Testing Async FastAPI With SQLAlchemy and pytest

Overview

Async test suites for FastAPI tend to fail in confusing ways: RuntimeError: Event loop is closed, Task attached to a different loop, data leaking from one test into the next, or async def functions are not natively supported because tests were silently skipped. The causes are usually the event loop lifetime and the database session lifetime.

This guide shows a setup where each test gets an isolated, rolled-back database state.

1. Install and Configure pytest-asyncio

pip install pytest pytest-asyncio httpx

# pytest.ini
[pytest]
asyncio_mode = auto

In auto mode, async def tests and async fixtures are handled without a marker on each. In strict mode (the default) you decorate tests with @pytest.mark.asyncio and async fixtures with @pytest_asyncio.fixture. Fixture loop-scope options have changed across pytest-asyncio versions, so check the documentation for yours.

2. Use httpx With ASGITransport

TestClient is sync and runs the app in its own thread and loop, which makes sharing an async engine awkward. For async tests use httpx.AsyncClient.

import httpx
from app.main import app


async def test_health():
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.get("/health")
    assert response.status_code == 200

Note that this does not run lifespan events by default. If your app relies on startup, use a lifespan manager or construct what you need in fixtures.

3. Engine and Per-Test Rollback

Create the engine once, create tables once, and give each test a connection with an outer transaction that is rolled back.

import pytest_asyncio
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine

from app.models import Base

TEST_DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/app_test"


@pytest_asyncio.fixture(scope="session")
async def engine():
    engine = create_async_engine(TEST_DATABASE_URL)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await engine.dispose()


@pytest_asyncio.fixture
async def session(engine):
    async with engine.connect() as conn:
        trans = await conn.begin()
        async with AsyncSession(
            bind=conn,
            join_transaction_mode="create_savepoint",
            expire_on_commit=False,
        ) as session:
            yield session
        await trans.rollback()

join_transaction_mode="create_savepoint" (SQLAlchemy 2.0) makes session.commit() inside the app release a savepoint rather than committing the outer transaction, so the final rollback undoes everything. On SQLAlchemy 1.4 the pattern differs, using event listeners to restart nested transactions; check the documentation for your version.

4. Override the Dependency

Point the app's session dependency at the test session.

import httpx
import pytest_asyncio

from app.deps import get_session
from app.main import app


@pytest_asyncio.fixture
async def client(session):
    async def override_get_session():
        yield session

    app.dependency_overrides[get_session] = override_get_session
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        yield client
    app.dependency_overrides.clear()

Always clear overrides so they do not leak to other tests. Now tests are short and isolated:

async def test_create_and_read_user(client):
    created = await client.post("/users", json={"name": "Asha"})
    assert created.status_code == 201

    fetched = await client.get(f"/users/{created.json()['id']}")
    assert fetched.json()["name"] == "Asha"

5. Fixing Event Loop Errors

The usual cause is an async resource (an engine, a connection pool) created in one event loop and used in another. If you create the engine at import time in your app module and each test runs in a fresh loop, pooled connections bound to an old loop fail with Event loop is closed.

Options:

• Keep session-scoped async fixtures and tests on the same loop scope. In recent pytest-asyncio versions this is controlled with loop scope settings; in older ones people overrode an event_loop fixture, which is deprecated and removed in newer versions. Follow the docs for your installed version rather than copying old snippets.
• In tests, use poolclass=NullPool on the test engine so no connection outlives a loop.
• Create the app engine in the lifespan or in a factory, not at import time, so tests can substitute theirs.

from sqlalchemy.pool import NullPool

engine = create_async_engine(TEST_DATABASE_URL, poolclass=NullPool)

6. Other Pitfalls

• If your code calls session.commit() and you test that data persisted, remember it persists only within the test transaction. A separate connection will not see it.
• Code that opens its own session (background tasks, scripts) bypasses your override. Route all session creation through the dependency or a factory you can patch.
• Run migrations in CI against the test database at least once, rather than relying only on create_all, so schema drift shows up.
• Use a real PostgreSQL for tests if you use PostgreSQL features; SQLite behaves differently for constraints, types and locking.

Conclusion

A reliable async test setup needs a consistent event loop scope, an httpx client with ASGI transport, a dependency override for the session, and a per-test transaction that is rolled back.

If your FastAPI test suite is flaky or slow and you want it restructured, I can help. Contact me.

Additional Resources

• FastAPI async tests
• FastAPI testing dependencies with overrides
• pytest-asyncio documentation