如何用SQLAlchemy将多个动态文本查询Union并绑定参数？

针对你遇到的循环查询效率低、Text对象无法用Union、参数绑定困难的问题，提供两种可行方案：

方案一：使用元组IN子句（推荐，性能更优） #

直接构造(col_1, col_2) IN (...)的条件查询，比Union更简洁，数据库优化器对这类语法的处理效率更高。

from sqlalchemy import text

# 提取所有(col_1, col_2)的元组对
value_pairs = [(item['col_1'], item['col_2']) for item in my_items]

# 构造带参数的SQL语句
sql = text("""
    SELECT col_1, col_2
    FROM my_table
    WHERE (col_1, col_2) IN :values
""")

# 执行查询，传入参数集合
result = conn.execute(sql, {"values": value_pairs})

注意：该语法需要数据库支持元组IN（PostgreSQL、MySQL 8.0+、SQLite 3.33.0+均支持）。

方案二：用SQLAlchemy Core构造Union查询 #

如果必须使用Union，不要用text()对象，改用Core的表结构构造子查询，自动处理参数绑定与语句兼容问题。

from sqlalchemy import select, union, Table, Column, MetaData

# 定义表结构（已有ORM模型可直接用`Model.__table__`替代）
metadata = MetaData()
my_table = Table(
    'my_table', metadata,
    Column('col_1', ...),  # 替换为实际字段类型
    Column('col_2', ...)
)

subqueries = []
params = {}
# 循环生成子查询并绑定唯一参数名
for idx, col_values in enumerate(my_items):
    param1 = f"c1_{idx}"
    param2 = f"c2_{idx}"
    # 构造带参数的子查询
    subq = select(my_table.c.col_1, my_table.c.col_2).where(
        (my_table.c.col_1 == params[param1]) &
        (my_table.c.col_2 == params[param2])
    )
    subqueries.append(subq)
    # 存入参数值
    params[param1] = col_values['col_1']
    params[param2] = col_values['col_2']

# 合并为Union查询
final_query = union(*subqueries)
# 执行查询并传入所有参数
result = conn.execute(final_query, params)

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