在 FastAPI 异步测试中访问测试数据库时出现 PendingRollbackError答案

【问题标题】：PendingRollbackError when accessing test database in FastAPI async test在 FastAPI 异步测试中访问测试数据库时出现 PendingRollbackError
【发布时间】：2022-05-06 00:30:54
【问题描述】：

我试图在 FastAPI 上运行测试时模仿 Django 行为：我想在每个测试开始时创建一个测试数据库，并在最后销毁它。问题是 FastAPI 的异步特性破坏了一切。当我进行健全性检查并将所有内容同步时，一切都运行良好。但是，当我尝试异步运行时，一切都会中断。这是我目前拥有的：

夹具：

@pytest.fixture(scope="session")
def event_loop():
    return asyncio.get_event_loop()


@pytest.fixture(scope="session")
async def session():
    sync_test_db = "postgresql://postgres:postgres@postgres:5432/test"
    if not database_exists(sync_test_db):
        create_database(sync_test_db)
    async_test_db = "postgresql+asyncpg://postgres:postgres@postgres:5432/test"
    engine = create_async_engine(url=async_test_db, echo=True, future=True)
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

    Session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
    async with Session() as session:
        def get_session_override():
            return session

        app.dependency_overrides[get_session] = get_session_override
        yield session
    drop_database(sync_test_db)

测试：

class TestSomething:
    @pytest.mark.asyncio
    async def test_create_something(self, session):
        data = {"some": "data"}
        response = client.post(
            "/", json=data
        )
        assert response.ok
        results = await session.execute(select(Something))  # <- This line fails
        assert len(results.all()) == 1

错误：

E                       sqlalchemy.exc.PendingRollbackError: This Session's transaction has been rolled back due to a previous exception during flush. To begin a new transaction with this Session, first issue Session.rollback(). Original exception was: Task <Task pending name='anyio.from_thread.BlockingPortal._call_func' coro=<BlockingPortal._call_func() running at /usr/local/lib/python3.9/site-packages/anyio/from_thread.py:187> cb=[TaskGroup._spawn.<locals>.task_done() at /usr/local/lib/python3.9/site-packages/anyio/_backends/_asyncio.py:629]> got Future <Future pending cb=[Protocol._on_waiter_completed()]> attached to a different loop (Background on this error at: https://sqlalche.me/e/14/7s2a)

/usr/local/lib/python3.9/site-packages/sqlalchemy/orm/session.py:601: PendingRollbackError

有什么想法我可能做错了吗？

【问题讨论】：

标签： python async-await sqlalchemy pytest fastapi

【解决方案1】：

在引发此错误之前，检查您的测试用例中涉及数据库的其他语句是否可能会失败。

对我来说，PendingRollbackError 是由先前测试引发的 InsertionError 引起的。

我所有的测试都是（异步）单元测试，涉及将数据库插入到 postgres 数据库中。测试后，数据库会话应该回滚其条目。

InsertionError 是由对数据库的插入导致唯一约束失败。随后的所有测试都提出了PendingRollbackError。

【讨论】：