SQL Alchemy调用SQL Server存储过程无执行，如何获取输出消息？

在Microsoft SQL Server Management Studio（Windows身份验证）中，无参数存储过程SCD_ENS_RATE_SHEET可正常执行，且已确认拥有执行权限，但使用SQL Alchemy调用该存储过程时，Python无报错却未执行数据库操作（该存储过程包含截断表等修改操作）。需要在Python代码中获取SQL Server的输出消息以排查问题，当前连接配置及执行代码如下：

连接配置代码 #

from sqlalchemy import create_engine
import urllib

MYDATABASE_params = 'DRIVER={SQL Server};' \
         'SERVER=MYSERVER;' \
         'PORT=1234;' \
         'DATABASE=MYDATABASE;' \
         'Trusted_Connection=yes;'

MYDATABASE_params = urllib.parse.quote_plus(MYDATABASE_params)

MYDATABASE_engine = create_engine('mssql+pyodbc:///?odbc_connect=%s' % MYDATABASE_params, echo=True)

存储过程执行代码 #

archive_trig = "EXEC SCD_ENS_RATE_SHEET"
MYDATABASE_engine.execute(archive_trig)

已排查：在Python中执行sp_who2时，结果与SSMS中基本一致，仅ProgramName列显示为Python，怀疑存在权限或事务提交问题。

1. 直接使用pyodbc捕获SQL Server消息 #

SQLAlchemy的默认execute方法不会主动捕获SQL Server的打印消息、警告或内部错误，改用pyodbc直接连接并绑定输出处理器：

import pyodbc
import urllib

# 构建ODBC连接字符串
conn_str = 'DRIVER={SQL Server};SERVER=MYSERVER;PORT=1234;DATABASE=MYDATABASE;Trusted_Connection=yes;'
conn = pyodbc.connect(conn_str)

# 定义消息处理函数，打印所有SQL Server输出
def sql_server_output_handler(message_type, message):
    print(f"[SQL Server消息] 类型:{message_type} 内容:{message}")

# 绑定输出处理器
conn.add_output_handler(sql_server_output_handler)

# 执行存储过程并处理事务
cursor = conn.cursor()
try:
    # 可选：添加SET NOCOUNT ON避免结果集干扰
    cursor.execute("SET NOCOUNT ON; EXEC SCD_ENS_RATE_SHEET")
    conn.commit()  # 关键：修改类操作必须提交事务，否则会自动回滚
    print("存储过程执行完成，已提交事务")
except Exception as e:
    print(f"执行异常: {str(e)}")
    conn.rollback()
finally:
    cursor.close()
    conn.close()

2. 在SQLAlchemy中获取底层连接捕获消息 #

如果需要保留SQLAlchemy的封装，可获取其底层的pyodbc连接来绑定输出处理器：

from sqlalchemy import create_engine
import urllib

MYDATABASE_params = 'DRIVER={SQL Server};' \
         'SERVER=MYSERVER;' \
         'PORT=1234;' \
         'DATABASE=MYDATABASE;' \
         'Trusted_Connection=yes;'

MYDATABASE_params = urllib.parse.quote_plus(MYDATABASE_params)
engine = create_engine('mssql+pyodbc:///?odbc_connect=%s' % MYDATABASE_params, echo=True)

def sql_server_output_handler(message_type, message):
    print(f"[SQL Server输出] {message}")

with engine.connect() as conn:
    # 获取底层pyodbc连接
    pyodbc_conn = conn.connection
    pyodbc_conn.add_output_handler(sql_server_output_handler)
    
    try:
        # 执行存储过程并提交事务
        result = conn.execute("SET NOCOUNT ON; EXEC SCD_ENS_RATE_SHEET")
        conn.commit()
        # 若存储过程有返回结果，可遍历输出
        for row in result:
            print(f"存储过程返回结果: {row}")
    except Exception as e:
        print(f"执行出错: {str(e)}")
        conn.rollback()

关键排查点 #

• 事务提交：SQLAlchemy默认处于自动提交关闭状态，所有修改类操作必须显式调用commit()，否则连接关闭时会自动回滚，这是"未执行操作"的最常见原因。
• 身份验证确认：在存储过程开头添加PRINT '当前执行用户: ' + SUSER_NAME();，通过Python捕获输出，对比SSMS中的执行用户是否一致，确认权限是否匹配。
• 存储过程内部错误：若存储过程包含TRY/CATCH块，需确保内部错误被抛出（如使用THROW语句），否则Python无法感知到错误。

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