如何在SQLAlchemy中运行多条依赖SQL语句？

针对你遇到的问题，核心原因是SQLAlchemy默认不处理多结果集批处理，以及单批执行外的变量无法跨会话批共享。以下是两种可靠的解决方式：

方式一：合并所有语句为单批，启用多结果集支持 #

将所有TSQL语句放在同一个文本块中，通过text()包裹并启用multiple_resultsets=True选项，让SQLAlchemy正确处理多语句的结果返回。同时避免用f-string直接拼接SQL值，改用参数化方式传递用户定义表类型（UDT）参数，杜绝SQL注入风险。

from sqlalchemy import text
from sqlalchemy.dialects.mssql import UDTT

# 1. 准备UDT参数（对应你的ObligorIDListType类型）
udt_type = UDTT(f"{schema}.ObligorIDListType")
# 转换为UDT要求的格式：列表内嵌套元组
udt_data = [(id,) for id in obligor_ids]

# 2. 编写完整SQL脚本（无需USE，通过execution_options切换数据库）
sql_script = text(f"""
    DECLARE @ObligorIDs AS {schema}.ObligorIDListType;
    INSERT INTO @ObligorIDs (ID) VALUES (:udt_values);
    SELECT * FROM {schema}.GetLUsByObligorID(@ObligorIDs);
""")

# 3. 执行脚本并处理结果
with engine.connect().execution_options(database=database) as conn:
    # 启用多结果集支持，处理INSERT和SELECT两个语句的结果
    result_proxy = conn.execute(sql_script, {"udt_values": udt_data}).execution_options(multiple_resultsets=True)
    # 遍历结果集，获取最后一个（SELECT语句的结果）
    for rs in result_proxy:
        final_results = rs.fetchall()
    conn.commit()

关键说明： #

• 用execution_options(database=database)替代USE语句，避免额外的批处理分割
• multiple_resultsets=True让SQLAlchemy能够处理同一个文本块内的多个语句结果
• 使用UDTT类型传递用户定义表参数，避免直接拼接SQL值导致注入风险

方式二：用存储过程封装逻辑（更推荐） #

如果你的业务逻辑固定，将整个流程封装为SQL Server存储过程，然后在SQLAlchemy中调用，这是更简洁且易维护的方式：

步骤1：创建存储过程 #

CREATE PROCEDURE {schema}.GetLUsByObligorIDs
    @ObligorIDs {schema}.ObligorIDListType READONLY
AS
BEGIN
    SELECT * FROM {schema}.GetLUsByObligorID(@ObligorIDs);
END

步骤2：SQLAlchemy调用存储过程 #

from sqlalchemy import text
from sqlalchemy.dialects.mssql import UDTT

udt_type = UDTT(f"{schema}.ObligorIDListType")
udt_data = [(id,) for id in obligor_ids]

with engine.connect().execution_options(database=database) as conn:
    result = conn.execute(
        text(f"EXEC {schema}.GetLUsByObligorIDs @ObligorIDs = :udt_values;"),
        {"udt_values": udt_data}
    )
    final_results = result.fetchall()
    conn.commit()

为什么之前的方式失败？ #

1. 一次性执行失败：默认情况下SQLAlchemy不处理多结果集批处理，USE语句会触发一个独立批，后续语句的结果无法被正确捕获，启用multiple_resultsets=True即可解决。
2. 分批次执行失败：每个session.execute()调用对应一个独立的TSQL批处理，变量@ObligorIDs仅在定义它的批内有效，跨批无法访问。

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