调用db.commit()触发RuntimeError: Event loop is closed问题修复求助

针对调用db.commit()时触发的事件循环关闭错误，以下是具体排查与修复方案：

核心原因分析 #

该错误本质是异步会话（AsyncSession）与事件循环生命周期不匹配，或是会话未正确绑定到测试用数据库引擎，导致事件循环已关闭后仍尝试执行数据库操作。

具体修复步骤 #

1. 修正db fixture，确保会话绑定测试引擎 #

原db fixture直接使用全局SessionLocal，可能未关联到测试专用的db_engine。修改conftest.py中的db fixture：

@pytest.fixture
async def db(db_engine) -> AsyncIterable[AsyncSession]:
    # 为每个测试用例创建独立异步会话，绑定到测试引擎
    async_session = sessionmaker(
        bind=db_engine,
        class_=AsyncSession,
        expire_on_commit=False  # 避免commit后实例脱管
    )
    async with async_session() as session:
        yield session

2. 在db_ins中添加实例刷新操作 #

异步会话执行commit后，实例会脱离会话管理，后续操作易引发异常。修改db_ins fixture：

@pytest.fixture
async def db_ins(db: AsyncSession) -> AsyncIterable[DBInsert]:
    async def insert(factory: SQLAlchemyModelFactory, **kwargs: Any) -> Any:
        instance = factory.build(**kwargs)
        db.add(instance)
        await db.commit()
        await db.refresh(instance)  # 刷新实例，重新绑定会话
        return instance
    yield insert

3. 显式指定测试会话的事件循环 #

若为pytest-asyncio事件循环管理问题，在conftest.py中添加全局事件循环fixture：

import asyncio

@pytest.fixture(scope="session")
def event_loop():
    # 创建独立会话级事件循环，避免默认循环提前关闭
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()

4. 全局SessionLocal定义验证（若有） #

若项目中存在全局SessionLocal定义，需确保测试时它能正确绑定到测试引擎；或直接采用步骤1的方式动态创建会话，避免依赖全局SessionLocal。

验证修复 #

修改完成后重新运行测试，test_create_user_failed_unique_phone_number及其他使用admin_user fixture的测试应能正常执行，不再触发事件循环关闭错误。

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