如何在SQLAlchemy中从自定义TextClause查询中获取列名

我开发的应用常规接收SQLAlchemy的selectable对象作为输入，调用reflection.Inspector.from_engine(engine).get_columns(selectable.name, schema=selectable.schema)获取列信息供给下游逻辑使用。
目前应用支持用户传入自定义TextClause查询作为输入，我需要实现和selectable对象一致的列名解析能力，是否可以从TextClause中逆向提取查询对应的列名？

已尝试的操作 #

>>> import sqlalchemy as sa
>>> query = "SELECT col_1, col_2 FROM table"
>>> selectable = sa.text(query)

# 以上步骤无法修改

>>> type(selectable)
<class 'sqlalchemy.sql.elements.TextClause'>
>>> connection_string = "postgresql+psycopg2://<user>:<password>@localhost:5432/<db>"
>>> engine = sa.create_engine(connection_string)
>>> columns = sa.engine.reflection.Inspector.from_engine(engine).get_columns(selectable, schema=None)

报错信息 #

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/Users/user/opt/anaconda3/envs/env/lib/python3.7/site-packages/sqlalchemy/engine/reflection.py", line 498, in get_columns
    conn, table_name, schema, info_cache=self.info_cache, **kw
  File "<string>", line 2, in get_columns
  File "/Users/user/opt/anaconda3/envs/env/lib/python3.7/site-packages/sqlalchemy/engine/reflection.py", line 55, in cache
    ret = fn(self, con, *args, **kw)
  File "/Users/user/opt/anaconda3/envs/env/lib/python3.7/site-packages/sqlalchemy/dialects/postgresql/base.py", line 3578, in get_columns
    connection, table_name, schema, info_cache=kw.get("info_cache")
  File "<string>", line 2, in get_table_oid
  File "/Users/user/opt/anaconda3/envs/env/lib/python3.7/site-packages/sqlalchemy/engine/reflection.py", line 55, in cache
    ret = fn(self, con, *args, **kw)
  File "/Users/user/opt/anaconda3/envs/env/lib/python3.7/site-packages/sqlalchemy/dialects/postgresql/base.py", line 3457, in get_table_oid
    raise exc.NoSuchTableError(table_name)
sqlalchemy.exc.NoSuchTableError: SELECT col_1, col_2 FROM table

你遇到的报错原因是Inspector.get_columns方法仅接受表名/视图名作为入参，你直接传入完整的TextClause对象会被当成表名去检索，自然会抛出不存在的错误。要提取TextClause对应查询的列信息，推荐使用以下两种方案：

方案1：利用数据库结果集元数据（生产环境首选，兼容性最高） #

不需要实际执行查询拉取全量数据，仅通过LIMIT 0的空查询即可拿到结果集的列元数据，完全兼容任意合法的自定义SQL，不管是多表关联、函数计算、字段别名等复杂场景都能正确返回列信息，数据库执行开销极低：

import sqlalchemy as sa

# 原有不可修改的逻辑
query = "SELECT col_1, col_2 FROM table"
selectable = sa.text(query)

connection_string = "postgresql+psycopg2://<user>:<password>@localhost:5432/<db>"
engine = sa.create_engine(connection_string)

# 新增类型判断分支
if isinstance(selectable, sa.sql.elements.TextClause):
    with engine.connect() as conn:
        # 套一层子查询加LIMIT 0，不会返回实际业务数据
        result = conn.execute(sa.select(sa.text("*")).select_from(selectable).limit(0))
        # 直接获取列名列表
        column_names = result.keys()
        # 如果需要和原有get_columns返回格式一致（包含字段类型、是否可为空等属性），可通过cursor.description解析
        full_column_info = []
        for col in result.cursor.description:
            col_type = conn.dialect.type_compiler.process(conn.dialect.get_description_type(col))
            full_column_info.append({
                "name": col[0],
                "type": col_type,
                "nullable": col[6]
            })

方案2：SQLAlchemy 2.0+ 离线解析方案 #

如果你使用的是2.0及以上版本的SQLAlchemy，可以通过方言编译能力离线解析SQL获取列信息，不需要连接数据库，但对SQL语法的兼容性低于方案1，部分特殊写法可能解析失败：

from sqlalchemy.sql import text
from sqlalchemy.dialects import postgresql

query = "SELECT col_1, col_2 FROM table"
selectable = text(query)

dialect = postgresql.dialect()
compiled = selectable.compile(dialect=dialect)
column_names = [col.name for col in compiled.statement.columns]

内容的提问来源于stack exchange，提问作者Pierre Delecto