SQLAlchemy抛出ProgrammingError时如何保留已中止事务？

嘿，这个场景我在维护遗留项目的集成测试套件时真的踩过坑！首先得明确：当SQLAlchemy抛出ProgrammingError时，不是SQLAlchemy主动中止事务，而是底层数据库连接进入了无效状态——大多数数据库（比如PostgreSQL、MySQL）在遇到这类严重错误（语法错误、约束违反、死锁等）后，会标记当前事务为不可用，必须回滚或断开重连才能继续操作。

不过我们还是有办法让测试用的事务套件继续工作，下面是几个实用的方案：

1. 捕获错误后重置连接与事务 #

如果你的测试fixture是基于函数级别的，可以在捕获到ProgrammingError时，手动重置数据库连接并重新开启事务，这样后续测试还能复用fixture的会话：

from sqlalchemy.exc import ProgrammingError
import pytest

@pytest.fixture(scope="function")
def test_session():
    session = create_your_session()  # 替换成你的会话创建逻辑
    session.begin()
    try:
        yield session
    except ProgrammingError:
        # 先回滚当前已失效的事务
        session.rollback()
        # 重置底层连接，清除无效状态
        session.connection().invalidate()
        # 重新开启一个干净的事务
        session.begin()
        # 把错误抛出去，让测试标记为失败
        raise
    finally:
        # 不管测试成功失败，最终回滚事务，避免污染数据库
        session.rollback()
        session.close()

2. 使用嵌套事务（Savepoints）隔离测试错误 #

如果想让单个测试的错误不影响整个fixture的顶层事务，可以用SQLAlchemy的嵌套事务（Savepoint）。每个测试运行在一个savepoint里，出错时只回滚这个savepoint，顶层事务依然保持活跃：

@pytest.fixture(scope="module")
def module_session():
    # 模块级别的顶层事务，整个模块的测试共用
    session = create_your_session()
    session.begin()
    yield session
    session.rollback()
    session.close()

@pytest.fixture(scope="function")
def test_session(module_session):
    # 每个测试创建一个嵌套事务（savepoint）
    savepoint = module_session.begin_nested()
    try:
        yield module_session
    except Exception:
        # 出错时只回滚到savepoint，不影响顶层事务
        savepoint.rollback()
        raise
    finally:
        # 测试成功也回滚savepoint，清除测试数据
        savepoint.rollback()

这个方案特别适合遗留代码——因为顶层事务一直保持，不用频繁重建会话，测试运行速度更快，而且单个测试的错误不会中断整个测试模块的执行。

3. 注意事项 #

• 不是所有数据库都支持嵌套事务：比如SQLite对savepoint的支持有限，如果你用的是SQLite，可能还是得用方案1。
• 遗留代码中的手动事务操作：如果你的遗留代码里有手动commit或rollback，会和fixture的事务逻辑冲突，需要先梳理原有代码的事务边界。
• 错误类型区分：有些ProgrammingError是不可恢复的（比如数据库连接断开），这时候重置连接也没用，这种情况可以让fixture直接抛出错误，终止测试。

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