如何在多个pytest中复用SQLAlchemy异步会话并每次测试后回滚事务？

错误信息翻译 #

无法在上下文管理器内操作已关闭的事务。请先完成上下文管理器，再执行后续命令。

问题根源 #

1. sessionmaker参数错误：直接将connection作为位置参数传入sessionmaker，未通过bind关键字明确指定连接，导致会话与连接的绑定逻辑异常
2. 嵌套事务重复操作：函数级夹具中，async with setup_session.begin_nested()上下文管理器会自动处理事务生命周期，手动调用await transaction.rollback()会触发对已关闭事务的重复操作
3. 会话初始化逻辑问题：模块级会话的创建方式未正确关联connection的事务上下文

修正后的代码 #

@pytest_asyncio.fixture(scope="module")
async def setup_session(engine: AsyncEngine) -> AsyncIterator[AsyncSession]:
    connection = await engine.connect()
    # 启动模块级事务，保持活跃至所有测试完成
    await connection.begin()

    # 正确配置sessionmaker，通过bind关键字指定连接
    async_session_maker = sessionmaker(
        bind=connection,
        class_=AsyncSession,
        expire_on_commit=False,
    )

    async_session = async_session_maker()
    try:
        yield async_session
    finally:
        await async_session.close()
        # 回滚模块级事务，清理所有测试数据
        await connection.rollback()
        await connection.close()


@pytest_asyncio.fixture(scope="function")
async def session(setup_session: AsyncSession) -> AsyncIterator[AsyncSession]:
    # 启动嵌套事务，上下文管理器自动处理回滚/提交
    async with setup_session.begin_nested():
        yield setup_session
    # 无需手动回滚，async with会自动回滚未提交的嵌套事务

关键说明 #

• 模块级夹具：创建长期活跃的数据库连接与事务，所有测试共用同一会话实例，模块测试完成后统一回滚并关闭连接，实现环境仅初始化一次
• 函数级夹具：每次测试启动嵌套事务，测试结束后上下文管理器自动回滚该事务，确保测试间状态完全隔离
• expire_on_commit=False：避免会话对象在事务操作后自动过期，防止测试中出现对象状态异常

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