Типы Python для типов SQL:
Я боролся с проблемой создания таблиц SQL на лету с типами sql по умолчанию. В итоге я получил следующие удобные функции для всех моих потребностей преобразования типа python в тип sql. Переход от типа sql к типу python тривиален, как будет объяснено в следующем разделе.
import sqlalchemy
import numpy as np
import datetime
import decimal
_type_py2sql_dict = {
int: sqlalchemy.sql.sqltypes.BigInteger,
str: sqlalchemy.sql.sqltypes.Unicode,
float: sqlalchemy.sql.sqltypes.Float,
decimal.Decimal: sqlalchemy.sql.sqltypes.Numeric,
datetime.datetime: sqlalchemy.sql.sqltypes.DateTime,
bytes: sqlalchemy.sql.sqltypes.LargeBinary,
bool: sqlalchemy.sql.sqltypes.Boolean,
datetime.date: sqlalchemy.sql.sqltypes.Date,
datetime.time: sqlalchemy.sql.sqltypes.Time,
datetime.timedelta: sqlalchemy.sql.sqltypes.Interval,
list: sqlalchemy.sql.sqltypes.ARRAY,
dict: sqlalchemy.sql.sqltypes.JSON
}
def type_py2sql(pytype):
'''Return the closest sql type for a given python type'''
if pytype in _type_py2sql_dict:
return _type_py2sql_dict[pytype]
else:
raise NotImplementedError(
"You may add custom `sqltype` to `"+str(pytype)+"` assignment in `_type_py2sql_dict`.")
def type_np2py(dtype=None, arr=None):
'''Return the closest python type for a given numpy dtype'''
if ((dtype is None and arr is None) or
(dtype is not None and arr is not None)):
raise ValueError(
"Provide either keyword argument `dtype` or `arr`: a numpy dtype or a numpy array.")
if dtype is None:
dtype = arr.dtype
#1) Make a single-entry numpy array of the same dtype
#2) force the array into a python 'object' dtype
#3) the array entry should now be the closest python type
single_entry = np.empty([1], dtype=dtype).astype(object)
return type(single_entry[0])
def type_np2sql(dtype=None, arr=None):
'''Return the closest sql type for a given numpy dtype'''
return type_py2sql(type_np2py(dtype=dtype, arr=arr))
Некоторые варианты использования:
>>> sqlalchemy.Column(type_py2sql(int))
Column(None, BigInteger(), table=None)
>>> type_py2sql(type('hello'))
sqlalchemy.sql.sqltypes.Unicode
>>> type_np2sql(arr=np.array([1.,2.,3.]))
sqlalchemy.sql.sqltypes.Float
Как я выбирал свой конверсионный набор:
Что я сделал, так это сопоставил все типы sql с их эквивалентными типами python. Затем я напечатал, какой тип python соответствует какому типу sql, и выбрал лучший тип sql для каждого типа python. Вот код, который я использовал для создания этого сопоставления:
#********** SQL to Python: one to one **********
type_sql2py_dict = {}
for key in sqlalchemy.types.__dict__['__all__']:
sqltype = getattr(sqlalchemy.types, key)
if 'python_type' in dir(sqltype) and not sqltype.__name__.startswith('Type'):
try:
typeinst = sqltype()
except TypeError as e: #List/array wants inner-type
typeinst = sqltype(None)
try:
type_sql2py_dict[sqltype] = typeinst.python_type
except NotImplementedError:
pass
#********** Python to SQL: one to many **********
type_py2sql_dict = {}
for key, val in type_sql2py_dict.items():
if not val in type_py2sql_dict:
type_py2sql_dict[val] = [key]
else:
type_py2sql_dict[val].append(key)
А вот вывод type_py2sql_dict для sqlalchemy версии 1.3.5:
{int: [sqlalchemy.sql.sqltypes.INTEGER,
sqlalchemy.sql.sqltypes.BIGINT,
sqlalchemy.sql.sqltypes.SMALLINT,
sqlalchemy.sql.sqltypes.Integer,
sqlalchemy.sql.sqltypes.SmallInteger,
sqlalchemy.sql.sqltypes.BigInteger],
str: [sqlalchemy.sql.sqltypes.CHAR,
sqlalchemy.sql.sqltypes.VARCHAR,
sqlalchemy.sql.sqltypes.NCHAR,
sqlalchemy.sql.sqltypes.NVARCHAR,
sqlalchemy.sql.sqltypes.TEXT,
sqlalchemy.sql.sqltypes.Text,
sqlalchemy.sql.sqltypes.CLOB,
sqlalchemy.sql.sqltypes.String,
sqlalchemy.sql.sqltypes.Unicode,
sqlalchemy.sql.sqltypes.UnicodeText,
sqlalchemy.sql.sqltypes.Enum],
float: [sqlalchemy.sql.sqltypes.FLOAT,
sqlalchemy.sql.sqltypes.REAL,
sqlalchemy.sql.sqltypes.Float],
decimal.Decimal: [sqlalchemy.sql.sqltypes.NUMERIC,
sqlalchemy.sql.sqltypes.DECIMAL,
sqlalchemy.sql.sqltypes.Numeric],
datetime.datetime: [sqlalchemy.sql.sqltypes.TIMESTAMP,
sqlalchemy.sql.sqltypes.DATETIME,
sqlalchemy.sql.sqltypes.DateTime],
bytes: [sqlalchemy.sql.sqltypes.BLOB,
sqlalchemy.sql.sqltypes.BINARY,
sqlalchemy.sql.sqltypes.VARBINARY,
sqlalchemy.sql.sqltypes.LargeBinary,
sqlalchemy.sql.sqltypes.Binary],
bool: [sqlalchemy.sql.sqltypes.BOOLEAN, sqlalchemy.sql.sqltypes.Boolean],
datetime.date: [sqlalchemy.sql.sqltypes.DATE, sqlalchemy.sql.sqltypes.Date],
datetime.time: [sqlalchemy.sql.sqltypes.TIME, sqlalchemy.sql.sqltypes.Time],
datetime.timedelta: [sqlalchemy.sql.sqltypes.Interval],
list: [sqlalchemy.sql.sqltypes.ARRAY],
dict: [sqlalchemy.sql.sqltypes.JSON]}
person
Simon Streicher
schedule
17.07.2019
