如何在SQLAlchemy中从方言特定类型获取通用数据类型？

Question

我想创建一个函数，将方言特定的数据库类型映射到通用的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列类型（String，Integer等），而不是特定于方言的列类型？

- Yuriy Stakhniak

---

重复：https://dev59.com/8b3pa4cB1Zd3GeqPg5Jn - Gord Thompson

@GordThompson，不幸的是，这里提供的解决方案（https://dev59.com/8b3pa4cB1Zd3GeqPg5Jn）对于特定于驱动程序的数据类型，如 cx_Oracle.DbType.DB_TYPE_VARCHAR 不起作用。 - MaxU - stand with Ukraine

@MaxU 关于“不适用于特定于驱动程序的数据类型”，如果您认为这是SQLAlchemy的.as_generic()实现的不足之处，请在GitHub上开启一个问题。 - Gord Thompson

@GordThompson，好的，在赏金过期后我们会尝试打开一个新的GitHub问题... - MaxU - stand with Ukraine

Answer 1

通过 此 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'>

- Gord Thompson