如何从 sqlalchemy 中的方言特定类型获取通用数据类型？答案

【问题标题】：How to get generic data type from dialect specific type in sqlalchemy?如何从 sqlalchemy 中的方言特定类型获取通用数据类型？
【发布时间】：2021-04-02 07:16:09
【问题描述】：

我想创建一个函数，将特定方言的数据库类型映射到通用 sqlalchemy 数据库类型。

例如：我正在尝试从 OracleDB 表中获取数据类型，然后使用收到的元数据在 MySql DB（或另一个）中创建一个表。我正在获取特定于方言的列类型。

这是创建 OracleDB 表的最少代码：

create table test_test(id int, num_int number(10), num_float number(10,4), dt date, ts timestamp, str varchar2(100), fl float);

insert into test_test(id) values(1);

接收数据库表列类型

from pprint import pprint
from sqlalchemy import sessionmaker, create_engine

engine = create_engine(...)
session = sessionmaker()
session.configure(bind=engine)
session = session()
res = session.execute('select * from test_test where 1 = 2')
columns = res.cursor.description
pprint(columns)

输出

[('ID', <cx_Oracle.DbType DB_TYPE_NUMBER>, 39, None, 38, 0, 1),
 ('NUM_INT', <cx_Oracle.DbType DB_TYPE_NUMBER>, 11, None, 10, 0, 1),
 ('NUM_FLOAT', <cx_Oracle.DbType DB_TYPE_NUMBER>, 16, None, 10, 4, 1),
 ('DT', <cx_Oracle.DbType DB_TYPE_DATE>, 23, None, None, None, 1),
 ('TS', <cx_Oracle.DbType DB_TYPE_TIMESTAMP>, 23, None, 0, 6, 1),
 ('STR', <cx_Oracle.DbType DB_TYPE_VARCHAR>, 100, 100, None, None, 1),
 ('FL', <cx_Oracle.DbType DB_TYPE_NUMBER>, 127, None, 126, -127, 1)]

如何获得通用的 sqlalchemy 列类型（字符串、整数等）而不是特定方言的列类型？

【问题讨论】：

stackoverflow.com/q/64214225/2144390 的副本
@GordThompson，不幸的是，here 提供的解决方案不适用于驱动程序特定的数据类型，例如 cx_Oracle.DbType.DB_TYPE_VARCHAR
@MaxU re：“不适用于特定于驱动程序的数据类型” - 如果您认为这是 SQLAlchemy 实现 .as_generic() 的缺陷，请打开 issue on GitHub。跨度>
@GordThompson，好的，我们会在赏金到期后尝试打开一个新的 GitHib 问题...

标签： python types sqlalchemy orm metadata

【解决方案1】：

通过this pull request 添加到SQLAlchemy 1.4 的.as_generic() 方法可以处理SQLAlchemy 本身在方言级别定义的Oracle 类型……

from sqlalchemy.dialects.oracle import VARCHAR
oracle_type = VARCHAR()
print(type(oracle_type))  # <class 'sqlalchemy.sql.sqltypes.VARCHAR'>
generic_type = oracle_type.as_generic()
print(type(generic_type))  # <class 'sqlalchemy.sql.sqltypes.String'>

...但它不适用于驱动程序级别的类型，除非驱动程序本身实现它：

from cx_Oracle import  DB_TYPE_VARCHAR
cx_oracle_type = DB_TYPE_VARCHAR
print(type(cx_oracle_type))  # <class 'cx_Oracle.DbType'>
generic_type = cx_oracle_type.as_generic()
# AttributeError: 'cx_Oracle.DbType' object has no attribute 'as_generic'

因此，除非 cx-Oracle 实现了这样的方法（值得怀疑），否则您可能必须滚动自己的函数才能做到这一点：

def cx_oracle_type_as_generic(type_):
    import sqlalchemy as sa
    if type_.name == "DB_TYPE_VARCHAR":
        return sa.sql.sqltypes.String()
    elif type_.name == "DB_TYPE_NVARCHAR":
        return sa.sql.sqltypes.Unicode()
    # … and so on …
    else:
        raise NotImplementedError


from cx_Oracle import  DB_TYPE_VARCHAR
cx_oracle_type = DB_TYPE_VARCHAR
print(type(cx_oracle_type))  # <class 'cx_Oracle.DbType'>
generic_type = cx_oracle_type_as_generic(cx_oracle_type)
print(type(generic_type))  # <class 'sqlalchemy.sql.sqltypes.String'>

【讨论】：