函数内执行SELECT原始SQL报错，单独执行正常求助

问题现象 #

单独执行包含SELECT语句的SQL代码时运行正常，但将其封装到函数中调用会抛出HY010函数序列错误，同一函数执行UPDATE语句可正常运行。

代码示例 #

单独执行代码 #

engine = create_engine("mssql+pyodbc://{}:{}@{}/{}?driver={}".format(
        username, password, server, database, driver))

with engine.connect() as conn:
    result = conn.execute(text(find_user))

    try:
        for row in result:
            print(row)
    except:
        # 省略异常处理逻辑
        pass

函数内执行代码 #

find_user = "SELECT * FROM UserTable where UserId = 001"

def db(sql_command):
    engine = create_engine("mssql+pyodbc://{}:{}@{}/{}?driver={}".format(
            username, password, server, database, driver), echo=True)

    with engine.begin() as conn:
        result = conn.execute(text(sql_command))
        print(result.rowcount)
        return result

user = db(find_user)
for row in user:
    print(row)

报错信息 #

(pyodbc.Error) ('HY010', '[HY010] [Microsoft][SQL Server Native Client 11.0]Function sequence error (0) (SQLFetch)')

已尝试的方法 #

• 使用.fetchall()但未解决问题
• 查阅SQLAlchemy官方文档、网络搜索及咨询AI均未找到有效方案

问题原因 #

engine.begin()会创建事务上下文，当函数执行到with块结束时，事务自动提交/回滚，连接被关闭并归还至连接池。此时返回的result对象（底层为pyodbc游标）依赖的连接已失效，外部遍历result时会触发函数序列错误。

而单独执行时使用engine.connect()，遍历操作在with块内部完成，连接仍处于打开状态，因此可以正常获取结果。UPDATE语句无需返回结果，即使连接关闭也不会触发错误。

解决方案 #

方案1：在事务上下文内获取所有结果后返回 #

在with块内调用.fetchall()将结果转为独立的列表对象（不依赖数据库连接）后返回：

find_user = "SELECT * FROM UserTable where UserId = 001"  # 若UserId为字符串类型需添加单引号

def db(sql_command):
    engine = create_engine("mssql+pyodbc://{}:{}@{}/{}?driver={}".format(
            username, password, server, database, driver), echo=True)

    with engine.begin() as conn:
        result = conn.execute(text(sql_command))
        print(result.rowcount)
        return result.fetchall()  # 在连接关闭前获取所有结果

user = db(find_user)
for row in user:
    print(row)

方案2：将结果遍历逻辑放在函数内部的事务上下文里 #

无需返回result对象，直接在with块内完成结果处理：

find_user = "SELECT * FROM UserTable where UserId = 001"

def db(sql_command):
    engine = create_engine("mssql+pyodbc://{}:{}@{}/{}?driver={}".format(
            username, password, server, database, driver), echo=True)

    with engine.begin() as conn:
        result = conn.execute(text(sql_command))
        print(result.rowcount)
        for row in result:
            print(row)

db(find_user)

方案3：非事务场景改用engine.connect() #

如果不需要事务支持，替换engine.begin()为engine.connect()，同时确保在with块内完成结果读取：

find_user = "SELECT * FROM UserTable where UserId = 001"

def db(sql_command):
    engine = create_engine("mssql+pyodbc://{}:{}@{}/{}?driver={}".format(
            username, password, server, database, driver), echo=True)

    with engine.connect() as conn:
        result = conn.execute(text(sql_command))
        print(result.rowcount)
        return result.fetchall()

user = db(find_user)
for row in user:
    print(row)

注意事项 #

• 若UserId为字符串类型，SQL语句中需为值添加单引号（如UserId = '001'），避免类型匹配错误。
• 禁止将依赖数据库连接的Result对象带出上下文管理器，必须在连接有效时完成结果读取。

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