SQLAlchemy查询IN子句参数过多如何无需逐句改写实现分批传递

方案1：通用分批工具函数（改造成本最低） #

仅需封装一次通用逻辑，所有符合场景的查询只需调整1行代码即可完成适配，无需逐句手写循环：

def query_in_chunks(base_query, in_column, params, chunk_size=100):
    all_results = []
    for i in range(0, len(params), chunk_size):
        chunk_params = params[i:i+chunk_size]
        chunk_res = base_query.filter(in_column.in_(chunk_params)).all()
        all_results.extend(chunk_res)
    return all_results

调用示例，原有查询：

# 改造前
res = session.query(orm.table.name, orm.table.age).filter(orm.table.id.in_(params)).all()

改造后：

# 改造后，原有查询的其他过滤条件可完全保留
base_query = session.query(orm.table.name, orm.table.age)
res = query_in_chunks(base_query, orm.table.id, params)

方案2：SQLAlchemy原生IN参数优化（无需手动循环） #

SQLAlchemy 1.2及以上版本提供了expanding绑定参数特性，原生支持IN子句的动态参数展开，可避免生成不同结构的SQL语句导致的执行计划缓存失效，多数场景下参数量级在数千以内时无需手动分批，数据库也可正常走索引：

from sqlalchemy import bindparam

query = session.query(orm.table.name, orm.table.age).filter(
    orm.table.id.in_(bindparam('id_list', expanding=True))
)
res = query.params(id_list=params).all()

如果仍触发顺序扫描，可配合方案1的分批逻辑共同使用。

方案3：VALUES临时表JOIN（万级以上大参数场景性能最优） #

当IN参数量级过万时，用VALUES构造临时列表和业务表JOIN的性能远高于IN子句，PostgreSQL查询优化器基本不会触发顺序扫描：

from sqlalchemy import values, column, Integer

# 构造临时ID列表
tmp_id_table = values(
    column('id', Integer),
    name='tmp_ids'
).data([(pid,) for pid in params])

# 关联查询
res = session.query(orm.table.name, orm.table.age).join(
    tmp_id_table,
    orm.table.id == tmp_id_table.c.id
).all()

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