运行时错误：当我调用 db.comit() 时事件循环已关闭

Question

错误

FAILED tests/users/api/test_auth.py::test_create_user_failed_unique_phone_number - RuntimeError: Task <Task pending name='Task-22' coro=<test_create_user_failed_unique_phone_number() running at /app/tests/users/api/test_auth.py:36> cb=[_run_until_complete_cb() at /usr/local/lib/python3.12/asyncio/base_events...

当我尝试调用 db.commit 时，总是发生错误。对于数据库连接，我使用数据库 url postgresql+asyncpg。

# conftest.py


pytest_plugins = [
    "tests.users.factories.users",
]

async def init_db(engine):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

async def close_db(engine):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

@pytest.fixture(scope="session", autouse=True)
async def db_engine():
    engine = create_async_engine(settings.DATABASE_URL)
    await init_db(engine)
    yield engine
    await close_db(engine)
    await engine.dispose()

@pytest.fixture
async def db() -> AsyncIterable[AsyncSession]:
    async with SessionLocal() as session:
        yield session

@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()
        return instance
    yield insert

@pytest.fixture(scope="session")
async def client() -> AsyncIterable[AsyncClient]:
    async with AsyncClient(app=app, base_url="http://testserver") as client:
        yield client

# tests.users.factories.users

class UserFactory(SQLAlchemyModelFactory):
    class Meta:
        model = Users
        abstract = True

    first_name = factory.Faker("first_name")
    last_name = factory.Faker("last_name")
    email = factory.Faker("email")
    phone = factory.Sequence(lambda n: fake_phone_number())
    is_active = True

    hash_password: str = factory.LazyAttribute(lambda _: PasswordHelper().hash("12345678"))

class AdminUserFactory(UserFactory):
    role = Role.ADMIN

@pytest.fixture
async def admin_user(db_ins: DBInsert) -> Users:
    user = await db_ins(AdminUserFactory)
    return user

当我使用 admin_user 夹具时，我收到 RuntimeError: Event Loop is close 异常。当我们在 db_ins 夹具中调用 db.commit() 时，就会发生这种情况。

我想用数据库解决这个问题

Answer 1

这是最新版本的

pytest-asyncio

(https://github.com/pytest-dev/pytest-asyncio/issues/706) 的问题。 当您使用不同范围的灯具时会发生这种情况。 要使其正常工作，请使用

pytest-asyncio

版本 0.21.1 或更低版本