如何从 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 列类型（字符串、整数等）而不是特定于方言的列类型？

Answer 1

通过此拉取请求.as_generic()添加到 SQLAlchemy 1.4 的方法可以处理 SQLAlchemy 本身在方言级别 \xe2\x80\xa6 定义的 Oracle 类型

\n

from sqlalchemy.dialects.oracle import VARCHAR\noracle_type = VARCHAR()\nprint(type(oracle_type))  # <class \'sqlalchemy.sql.sqltypes.VARCHAR\'>\ngeneric_type = oracle_type.as_generic()\nprint(type(generic_type))  # <class \'sqlalchemy.sql.sqltypes.String\'>\n

\n

\xe2\x80\xa6 但它不适用于驱动程序级别类型，除非驱动程序本身实现它：

\n

from cx_Oracle import  DB_TYPE_VARCHAR\ncx_oracle_type = DB_TYPE_VARCHAR\nprint(type(cx_oracle_type))  # <class \'cx_Oracle.DbType\'>\ngeneric_type = cx_oracle_type.as_generic()\n# AttributeError: \'cx_Oracle.DbType\' object has no attribute \'as_generic\'\n

\n

因此，除非 cx-Oracle 实现了这样的方法（值得怀疑），否则您可能必须推出自己的函数来执行此操作：

\n

def cx_oracle_type_as_generic(type_):\n    import sqlalchemy as sa\n    if type_.name == "DB_TYPE_VARCHAR":\n        return sa.sql.sqltypes.String()\n    elif type_.name == "DB_TYPE_NVARCHAR":\n        return sa.sql.sqltypes.Unicode()\n    # \xe2\x80\xa6 and so on \xe2\x80\xa6\n    else:\n        raise NotImplementedError\n\n\nfrom cx_Oracle import  DB_TYPE_VARCHAR\ncx_oracle_type = DB_TYPE_VARCHAR\nprint(type(cx_oracle_type))  # <class \'cx_Oracle.DbType\'>\ngeneric_type = cx_oracle_type_as_generic(cx_oracle_type)\nprint(type(generic_type))  # <class \'sqlalchemy.sql.sqltypes.String\'>\n

\n