SQLAlchemy bindparam在Azure MSSQL失效但MySQL正常，求原因与解决

问题重现 #

代码示例 #

with engine.begin() as conn:
    print(f"Running with {engine.name=}")
    sql_q = text("SELECT count(*) FROM ticker WHERE id in :ids")
    sql_q.bindparams(bindparam('ids', expanding=True))

    result = conn.execute(
        sql_q, ids=[300, 400]
    ).scalar_one()
    print(f"{engine.name=}: {result=}")

MySQL环境运行结果 #

Running with engine.name='mysql'
engine.name='mysql': result=2

生成的查询语句：SELECT count(*) FROM ticker WHERE id in %(ids)s，参数：{'ids': [300, 400]}

Azure MSSQL环境报错信息 #

Running with engine.name='mssql'
sqlalchemy.exc.ProgrammingError: (pyodbc.ProgrammingError) ("A TVP's rows must be Sequence objects.", 'HY000')
[SQL: SELECT count(*) FROM ticker WHERE id in ?]
[parameters: ([300, 400],)]

环境版本 #

• Python 3.8.10
• SQLAlchemy==1.4.39
• pyodbc==4.0.39

原因分析 #

SQLAlchemy的bindparam('ids', expanding=True)在不同数据库的实现逻辑存在差异：

• MySQL：自动将列表参数展开为IN子句的多个占位符，适配MySQL的参数传递机制。
• Azure MSSQL：将展开参数解析为表值参数（TVP），而TVP要求每个参数元素必须是序列类型（如单元素元组(300,)）。直接传入整数列表[300,400]不符合TVP的格式要求，因此触发报错。

解决办法 #

方法1：适配MSSQL表值参数格式 #

将整数列表转换为单元素元组的列表，并调整SQL语句以提取TVP中的值：

with engine.begin() as conn:
    print(f"Running with {engine.name=}")
    sql_q = text("SELECT count(*) FROM ticker WHERE id IN (SELECT value FROM :ids)")
    sql_q.bindparams(bindparam('ids', expanding=True, type_=Integer))

    # 转换列表元素为单元素元组
    result = conn.execute(
        sql_q, ids=[(300,), (400,)]
    ).scalar_one()
    print(f"{engine.name=}: {result=}")

方法2：使用SQLAlchemy Core表达式（推荐） #

利用SQLAlchemy的抽象层自动适配不同数据库语法，无需手动处理差异：

from sqlalchemy import select, func, Table, Column, Integer, MetaData

# 定义表结构（已有ORM模型可直接复用）
metadata = MetaData()
ticker = Table('ticker', metadata, Column('id', Integer))

with engine.begin() as conn:
    print(f"Running with {engine.name=}")
    query = select(func.count()).where(ticker.c.id.in_([300, 400]))
    result = conn.execute(query).scalar_one()
    print(f"{engine.name=}: {result=}")

方法3：手动生成占位符 #

若坚持使用原生SQL文本，可手动生成与列表长度匹配的占位符并绑定参数（注意：需确保参数安全，避免SQL注入）：

with engine.begin() as conn:
    print(f"Running with {engine.name=}")
    ids = [300, 400]
    # 生成对应数量的占位符
    placeholders = ", ".join([f":id_{i}" for i in range(len(ids))])
    sql_q = text(f"SELECT count(*) FROM ticker WHERE id IN ({placeholders})")
    
    # 构造参数字典
    params = {f"id_{i}": val for i, val in enumerate(ids)}
    result = conn.execute(sql_q, params).scalar_one()
    print(f"{engine.name=}: {result=}")

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