如何让SQLAlchemy捕获网络中断时的pymssql异常？

这种情况并非你遗漏了配置，而是SQLAlchemy对pymssql驱动的异常捕获存在边缘场景的局限性，以下是具体原因和处理方案：

为什么抛出的是pymssql异常？ #

SQLAlchemy确实会将大部分底层驱动异常包装为自身的SQLAlchemyError子类（比如OperationalError），但这个包装逻辑主要覆盖通过SQLAlchemy Core/ORM执行的查询流程。如果你直接获取了pymssql的原生cursor并调用cursor.execute()，这个操作绕过了SQLAlchemy的异常拦截层，底层驱动的异常会直接向上抛出，不会被SQLAlchemy包装。

处理方案 #

1. 优先使用SQLAlchemy的官方执行方式 #

避免直接操作原生cursor，改用SQLAlchemy Core或ORM的API执行查询：

# 用SQLAlchemy Core执行
from sqlalchemy import text

with engine.connect() as conn:
    try:
        result = conn.execute(text("SELECT * FROM your_table"))
        # 处理结果
    except sqlalchemy.exc.OperationalError as e:
        # 捕获SQLAlchemy包装后的连接异常
        print(f"连接异常: {e}")

# 用ORM Session执行
from sqlalchemy.orm import Session

with Session(engine) as session:
    try:
        result = session.execute(text("SELECT * FROM your_table"))
        # 处理结果
    except sqlalchemy.exc.OperationalError as e:
        print(f"连接异常: {e}")

这种方式下，网络中断之类的连接异常会被SQLAlchemy包装为OperationalError，无需直接处理pymssql的异常。

2. 开启连接池预检查 #

在创建SQLAlchemy引擎时添加pool_pre_ping=True配置，让连接池在分配连接前自动检查连接有效性，提前淘汰失效连接，从根源减少这类异常的出现：

from sqlalchemy import create_engine

engine = create_engine(
    "mssql+pymssql://user:pass@host/db",
    pool_pre_ping=True
)

3. 若必须使用原生cursor #

如果业务场景要求必须操作原生cursor，只能手动捕获pymssql的异常，也可以自行将其包装为SQLAlchemy异常：

import pymssql
from sqlalchemy.exc import OperationalError

with engine.connect() as conn:
    # 获取原生cursor
    raw_cursor = conn.connection.cursor()
    try:
        raw_cursor.execute("SELECT * FROM your_table")
    except pymssql._mssql.MSSQLDatabaseException as e:
        # 手动包装为SQLAlchemy异常
        raise OperationalError(str(e), params=None, orig=e) from e

内容的提问来源于stack exchange，提问作者Phantom of Krankor