如何在Python+SQLAlchemy中捕获SQL Server的PRINT/RAISERROR消息？

要在Python（SQLAlchemy+pyodbc）中捕获SQL Server存储过程输出的PRINT或低级别RAISERROR消息，需要利用pyodbc的消息回调机制，以下是具体实现步骤：

核心思路 #

SQL Server的PRINT和严重级别≤10的RAISERROR属于信息性消息，不会触发Python的异常，但可以通过pyodbc提供的消息回调函数捕获。SQLAlchemy引擎底层基于pyodbc连接，因此可以获取原生连接并注册回调函数来处理这些消息。

实现代码 #

1. 导入依赖库 #

import sqlalchemy as sa
import pyodbc

2. 定义消息回调函数 #

这个函数会在收到SQL Server的信息消息时被调用，可根据需求处理消息（比如打印、存储到变量等）：

def sql_server_message_handler(conn, msg_text, msg_type, msg_state, msg_severity, msg_line, msg_server, msg_proc, msg_num):
    # 打印捕获到的消息，也可以将消息存入列表或其他存储结构
    print(f"[SQL Server消息] {msg_text}")
    # 返回0表示继续执行后续逻辑，返回1会终止当前操作
    return 0

3. 创建引擎并执行存储过程 #

通过SQLAlchemy获取底层pyodbc连接，注册回调后执行存储过程：

# 替换为你的SQL Server连接信息
connection_string = "mssql+pyodbc://用户名:密码@服务器名/数据库名?driver=ODBC+Driver+17+for+SQL+Server"
engine = sa.create_engine(connection_string)

with engine.connect() as conn:
    # 获取原生pyodbc连接对象
    raw_pyodbc_conn = conn.connection
    # 注册消息回调函数
    raw_pyodbc_conn.set_attr(pyodbc.SQL_ATTR_INFO_MESSAGE_HANDLER, sql_server_message_handler)
    
    # 执行存储过程
    conn.execute(sa.text("EXEC dbo.pyTest 1"))
    
    # 如果存储过程返回多个结果集，需遍历所有结果集确保消息被完全捕获
    while raw_pyodbc_conn.nextset():
        pass

print("存储过程执行完成")

注意事项 #

• RAISERROR严重级别区分：如果RAISERROR的严重级别>10（比如16），SQL Server会抛出异常，此时需要通过try-except捕获pyodbc.Error，而非消息回调。
• 结果集处理：若存储过程返回多个结果集，必须遍历所有结果集（如示例中的while raw_pyodbc_conn.nextset()），否则部分消息可能无法被触发捕获。
• 驱动版本：确保使用的ODBC Driver for SQL Server版本（推荐17+）和pyodbc版本支持SQL_ATTR_INFO_MESSAGE_HANDLER属性。

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