使用SQLAlchemy .in_()传入大量值时触发COUNT字段语法错误

问题 #

在Flask API项目中使用SQLAlchemy操作数据库，想要实现如下SQL查询：

SELECT * from trip_metadata where trip_id in ('trip_id_1', 'trip_id_2', ..., 'trip_id_n')

编写的代码如下：

trips_ids = ['trip_id_1', 'trip_id_2', ..., 'trip_id_n']
result = session.query(dal.trip_table).filter(dal.trip_table.columns.trip_id.in_(trips_ids)).all()

当n较小（如n=10）时代码运行正常，但当n>1000时抛出错误：

sqlalchemy.exc.DBAPIError: (pyodbc.Error) ('07002', '[07002] [Microsoft][ODBC Driver 17 for SQL Server]COUNT field incorrect or syntax error (0) (SQLExecDirectW)')')
[SQL: SELECT * FROM trip_metadata 
WHERE trip_metadata.trip_id IN (?, ?, ..., ?)]
[parameters: ('ABC12345-XXXX-XXXX-XXXX-000000000000', 'DEF12345-XXXX-XXXX-XXXX-000000000000', ..., 'GHI12345-XXXX-XXXX-XXXX-000000000000')]
(Background on this error at: https://sqlalche.me/e/14/dbapi)
127.0.0.1 - - [05/Jan/2023 10:35:48] "POST /api/v1/tripsAggregates HTTP/1.1" 500 -

使用原生SQL文本执行时，即使n很大也能正常运行：

from sqlalchemy import text
trip_ids_tuple = ('trip_id_1', 'trip_id_2', ..., 'trip_id_n')
result = session.execute(text(f"SELECT * FROM trip_metadata where trip_id in {trip_ids_tuple}"))

但希望保留SQLAlchemy的filter方法适配后续复杂查询，求解决办法。

解决方案 #

1. 拆分ID列表为小批次查询 #

SQL Server的ODBC驱动对参数数量有默认限制（通常为1000个），可以将大ID列表拆分为多个≤1000的子列表，分别查询后合并结果：

from itertools import islice

def chunked(iterable, size):
    iterator = iter(iterable)
    while chunk := list(islice(iterator, size)):
        yield chunk

trips_ids = ['trip_id_1', 'trip_id_2', ..., 'trip_id_n']
results = []
# 用999而非1000，预留余量避免边界问题
for chunk in chunked(trips_ids, 999):
    chunk_result = session.query(dal.trip_table).filter(dal.trip_table.columns.trip_id.in_(chunk)).all()
    results.extend(chunk_result)

2. 使用SQL Server表值参数（推荐） #

对于超大量ID的场景，表值参数是更高效且合规的方案：
首先在SQL Server中预先创建表值类型：

CREATE TYPE TripIdList AS TABLE (TripId VARCHAR(50) PRIMARY KEY)

然后在SQLAlchemy中实现关联查询：

from sqlalchemy import func

# 构造表值参数数据结构
tvp_data = [{'TripId': tid} for tid in trips_ids]
# 将数据传入表值参数并生成子查询
subquery = session.query(func.table_valued_parameter(tvp_data, 'TripIdList')).subquery()

# 通过关联查询获取目标数据
results = session.query(dal.trip_table).join(
    subquery,
    dal.trip_table.columns.trip_id == subquery.c.TripId
).all()

这种方式完全规避了参数数量限制，同时保留SQLAlchemy的ORM查询能力，适配后续复杂需求。

3. 调整ODBC驱动参数限制（不推荐） #

部分ODBC驱动支持通过配置调整最大参数数量，但该方式依赖环境和驱动版本，且可能引发其他潜在问题，不建议在生产环境使用。

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