PendingRollbackError when accessing test database in FastAPI async test

Question

I'm trying to mimic Django behavior when running tests on FastAPI: I want to create a test database in the beginning of each test, and destroy it in the end. The problem is the async nature of FastAPI is breaking everything. When I did a sanity check and turned everything synchronous, everything worked beautifully. When I try to run things async though, everything breaks. Here's what I have at the moment:

The fixture:

@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)

The test:

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

The error:

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

Any ideas what I might be doing wrong?

Answer 1

Check if other statements in your test-cases involving the database might fail before this error is raised.

For me the PendingRollbackError was caused by an InsertionError that was raised by a prior test.

All my tests were (async) unit tests that involved database insertions into a postgres database. After the tests, the database session was supposed to do a rollback of its entries.

The InsertionError was caused by Insertions to the database that failed a unique constraint. All subsequent tests raised the PendingRollbackError.