如何在SQLAlchemy中调用带TVP的SQL Server存储过程并获取多结果集？

不管用pyodbc还是pymssql驱动，都可以通过直接操作原始连接的cursor实现同时传递表值参数（TVP）和获取多个结果集，以下是具体方案：

方案一：使用pyodbc驱动 #

利用pyodbc原生对TVP的支持，结合cursor的nextset()方法遍历所有结果集：

from sqlalchemy import create_engine
import pyodbc

# 创建SQLAlchemy引擎
engine = create_engine("mssql+pyodbc://username:password@server/database?driver=ODBC+Driver+17+for+SQL+Server")

with engine.raw_connection() as conn:
    cursor = conn.cursor()
    # 定义表值参数：第一个参数是SQL Server中自定义表类型的全名，第二个是对应数据
    tvp = pyodbc.TableValuedParameter(
        "dbo.MyTableType",  # 替换为你的自定义表类型名
        [("foo", 1), ("bar", 2)]  # 数据需与表类型的列顺序、类型匹配
    )
    # 执行存储过程
    cursor.execute("{CALL testtypeprocedure (?)}", (tvp,))
    
    # 获取第一个结果集
    first_result = cursor.fetchall()
    print("第一个结果集:", first_result)
    
    # 遍历后续所有结果集
    while cursor.nextset():
        next_result = cursor.fetchall()
        print("后续结果集:", next_result)
    
    # 如需提交事务（若存储过程有写操作）
    cursor.commit()

关键说明： #

• 直接使用pyodbc的TableValuedParameter构造TVP，确保参数格式符合SQL Server要求
• 通过原始cursor的nextset()方法切换结果集，避免SQLAlchemy execute()只返回第一个结果集的限制

方案二：使用pymssql驱动 #

pymssql的callproc方法不支持TVP，但直接执行EXEC语句传递TVP数据列表即可，同样配合nextset()获取多结果集：

from sqlalchemy import create_engine

# 创建SQLAlchemy引擎
engine = create_engine("mssql+pymssql://username:password@server/database")

with engine.raw_connection() as conn:
    cursor = conn.cursor()
    # 准备TVP数据：需与自定义表类型的列顺序、类型严格匹配
    tvp_data = [("foo", 1), ("bar", 2)]
    # 执行存储过程，用%s作为参数占位符
    cursor.execute("EXEC testtypeprocedure @tvp = %s", (tvp_data,))
    
    # 获取第一个结果集
    first_result = cursor.fetchall()
    print("第一个结果集:", first_result)
    
    # 遍历后续所有结果集
    while cursor.nextset():
        next_result = cursor.fetchall()
        print("后续结果集:", next_result)
    
    # 提交事务（按需）
    cursor.commit()

关键说明： #

• 避免使用pymssql的callproc，改用直接执行EXEC语句的方式传递TVP
• TVP数据需以列表嵌套元组的形式传入，确保与SQL Server表类型定义一致

核心思路总结 #

两种方案的本质都是绕开SQLAlchemy高层API对结果集的限制，直接操作底层驱动的cursor：

1. 通过engine.raw_connection()获取原始数据库连接
2. 用驱动原生方式传递TVP（pyodbc用TableValuedParameter，pymssql直接传数据列表）
3. 执行存储过程后，通过cursor.nextset()遍历所有结果集

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