如何测试回滚后重试SQLAlchemy插入的FastAPI路由？

错误Can't operate on closed transaction inside context manager的核心原因是测试环境与生产环境的会话/事务生命周期不一致：

• 生产环境中，get_async_db_session为每个请求生成全新的会话，请求结束后会话自动销毁，路由内的rollback()仅影响当前会话的事务，不会干扰后续请求。
• 但测试里的db fixture是在一个持久化的连接事务上下文中创建会话，且整个测试函数复用同一个会话。当路由第一次请求执行await session.rollback()时，会直接关闭connection.begin()创建的顶层事务上下文，导致第二次请求时会话处于事务已关闭的不可用状态。

提供两种可行修改方案，任选其一即可：

方案一：修改测试Fixture，对齐生产环境会话逻辑 #

调整测试中的会话生成逻辑，让每个请求都获得全新的会话，和生产环境行为保持一致：

修改tests.py中的fixture代码：

# 保留原有的event_loop、db_engine fixture不变

@pytest.fixture(scope="function")
async def db_session_maker(db_engine):
    # 创建与生产环境一致的会话生成器
    session_maker = async_sessionmaker(bind=db_engine, class_=AsyncSession, expire_on_commit=False)
    yield session_maker

@pytest.fixture(scope="function")
async def client(db_session_maker):
    # 重写依赖，每次请求生成新会话
    async def override_get_async_db_session():
        async with db_session_maker() as session:
            yield session
    app.dependency_overrides[get_async_db_session] = override_get_async_db_session
    async with AsyncClient(app=app, base_url="http://test") as c:
        yield c

同时移除原来的db fixture。修改后每个请求都会生成独立会话，路由内的rollback()不会影响后续请求。

方案二：修改路由代码，用嵌套事务处理局部回滚 #

如果不想改动测试逻辑，可以调整路由的错误处理，使用嵌套事务隔离插入失败的回滚操作，避免关闭顶层事务：

修改main.py的路由函数：

@app.post("/add")
async def root(session=Depends(get_async_db_session)):
    for name in ("Max", "Cody", "Robby"):
        # 开启嵌套事务，仅回滚本次插入操作
        async with session.begin_nested():
            session.add(Animal(name=name))
            try:
                await session.flush()
            except IntegrityError:
                # 嵌套事务回滚不影响顶层事务
                continue

        await session.commit()
        return name

    return None

嵌套事务(begin_nested())会创建子事务，插入失败时仅回滚子事务，不会关闭顶层事务，测试中复用会话时也不会出现事务已关闭的报错。

修改后重新运行测试命令：

pytest --asyncio-mode=auto tests.py

两次请求均会正常执行，断言通过。

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