Testing with Async Database Fixtures in FastAPI

Key takeaways:

  • One session-scoped engine, one function-scoped transaction: build the schema once and undo each test.
  • Bind the session to an already-open connection and roll that connection's transaction back — never drop and recreate tables per test.
  • join_transaction_mode="create_savepoint" is what lets the endpoint call commit() without escaping your rollback.
  • Override the session dependency, and clear the overrides in teardown so nothing leaks between tests.
  • Match asyncio_default_fixture_loop_scope to the lifetime of your longest-lived async fixture.

This page applies the boundaries from Transaction Management and Rollback to a test suite, and sits under Async Database Sessions.

Nesting of the engine, transaction and session fixtures A session-scoped engine fixture contains a per-test connection and outer transaction, which contains the session bound to it and the app's savepoint. Rolling back the outer transaction discards everything the test wrote. Fixture nesting, outermost first engine (scope=session) - schema created once connection + outer transaction (per test) db_session bound to that connection endpoint commit() -> releases a SAVEPOINT Roll back the outer transaction and every layer above vanishes.
Isolation comes from the outer transaction, so tests never wait for tables to be dropped and rebuilt.

The Problem This Solves

Database tests fail in two characteristic ways. Either they are slow, because every test recreates the schema, or they are flaky, because they share state and the outcome depends on ordering. Both are solved by the same idea: give each test a transaction and never let it commit anything you cannot take back.

Doing that with an async session adds a second problem that has nothing to do with databases — event loops. An async fixture and the test that uses it must run on the same loop, and pytest-asyncio's defaults will happily give them different ones.

Why It Happens: Two Independent Lifetimes

An async SQLAlchemy engine owns a connection pool, and every pooled connection is created on some event loop. If a session-scoped engine fixture is created on loop A and a test then runs on loop B, the first query fails at the driver layer, not at the ORM layer, which makes the error message unhelpful.

pytest-asyncio governs this with two settings. asyncio_mode = auto makes every async def test runnable without a per-test marker. asyncio_default_fixture_loop_scope decides which loop async fixtures get. The rule to internalise: the loop must live at least as long as the longest-lived async fixture. A session-scoped engine therefore needs a session-scoped loop, and the tests using it need loop_scope="session" to match.

Without any plugin at all, the failure is unmistakable:

$ GET /pytest/no-asyncio-mode
200 OK
F                                                                        [100%]
=================================== FAILURES ===================================
____________________ test_without_pytest_asyncio_configured ____________________
async def functions are not natively supported.
You need to install a suitable plugin for your async framework, for example:
  - anyio
  - pytest-asyncio
  - pytest-tornasync
  - pytest-trio
  - pytest-twisted
=========================== short test summary info ============================
FAILED test_broken.py::test_without_pytest_asyncio_configured - Failed: async...
1 failed in 0.00s

That is a real pytest run against a file containing one trivially-passing async test and no plugin configuration. It fails not because the assertion is wrong but because nobody awaited the coroutine.

The Fix: Three Fixtures That Nest

Start with the configuration, because everything else depends on it:

[pytest]
asyncio_mode = auto
asyncio_default_fixture_loop_scope = session

Then the fixtures. Each one is deliberately narrow: the engine builds the schema, the session owns a rollback, and the client wires the app to that session.

import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.pool import StaticPool

from myapp import Base, app, get_session


@pytest_asyncio.fixture(scope="session")
async def engine():
    """One in-memory database for the whole session; schema created once."""
    engine = create_async_engine(
        "sqlite+aiosqlite://", poolclass=StaticPool, connect_args={"check_same_thread": False}
    )
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()


@pytest_asyncio.fixture
async def db_session(engine):
    """Bind the session to an OUTER transaction and roll it back, whatever the test committed."""
    async with engine.connect() as connection:
        transaction = await connection.begin()
        maker = async_sessionmaker(bind=connection, expire_on_commit=False, join_transaction_mode="create_savepoint")
        async with maker() as session:
            yield session
        await transaction.rollback()


@pytest_asyncio.fixture
async def client(db_session: AsyncSession):
    """Point the app's session dependency at the test's transaction, then undo the override."""
    app.dependency_overrides[get_session] = lambda: db_session
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as c:
        yield c
    app.dependency_overrides.clear()

Four decisions in there deserve explanation.

StaticPool with an in-memory URL. sqlite+aiosqlite:// creates a database that lives inside one connection. A normal pool would hand out a second, empty connection to the next caller and your tables would appear to vanish. StaticPool reuses the single connection so the whole suite sees one database.

Binding the sessionmaker to connection, not engine. This is what puts the session inside a transaction that the fixture, not the session, controls.

join_transaction_mode="create_savepoint". Without it, the application's commit() would end your outer transaction and there would be nothing left to roll back. With it, SQLAlchemy opens a savepoint for the session's work, so a commit() in the app resolves to releasing that savepoint. The data stays visible to the test and stays inside the outer transaction.

app.dependency_overrides.clear() after the yield. Overrides live on the app object, which outlives the test. Clearing in teardown is the difference between an isolated suite and a suite where test ordering matters.

What the Fixtures Guarantee

The tests below are written to prove the isolation rather than to exercise business logic. The app under test exposes POST /widgets (which flushes but never commits) and GET /widgets.

import pytest

pytestmark = pytest.mark.asyncio(loop_scope="session")


async def test_create_widget_returns_the_row(client):
    response = await client.post("/widgets", params={"name": "sprocket"})
    assert response.status_code == 200
    assert response.json() == {"id": 1, "name": "sprocket"}


async def test_previous_test_left_nothing_behind(client):
    """Proves the rollback: this is a separate test, so the sprocket is gone."""
    response = await client.get("/widgets")
    assert response.json() == []


async def test_ids_restart_because_the_insert_was_undone(client):
    created = await client.post("/widgets", params={"name": "gasket"})
    assert created.json()["id"] == 1


async def test_validation_failure_writes_nothing(client):
    bad = await client.post("/widgets", params={"name": "   "})
    assert bad.status_code == 422
    assert bad.json() == {"detail": "name must not be blank"}
    assert (await client.get("/widgets")).json() == []


async def test_the_session_fixture_and_the_endpoint_share_a_transaction(client, db_session):
    from sqlalchemy import select

    from myapp import Widget

    await client.post("/widgets", params={"name": "flange"})
    rows = (await db_session.execute(select(Widget))).scalars().all()
    assert [w.name for w in rows] == ["flange"]

Running that suite really does pass, in a real subprocess:

$ GET /pytest/db-fixtures
200 OK
.....                                                                    [100%]
5 passed in 0.00s

Each of those five dots is load-bearing:

  • The first test creates widget id 1.
  • The second sees an empty list. Nothing was dropped or truncated between them; the outer transaction was simply rolled back.
  • The third gets id 1 again, which is a stronger claim than "the row is gone" — the autoincrement sequence was rolled back too, so the tests are genuinely order-independent.
  • The fourth confirms a 422 leaves no residue, closing the loop with Transaction Management and Rollback.
  • The fifth is the one that proves the wiring: the test queries db_session directly and sees the row the endpoint inserted. If the override had failed, the app would have used its production engine and this assertion would find nothing.

That last test is the one to write first when adopting this pattern, because a broken override is otherwise invisible — the tests still pass, they just pass against the wrong database.

Adapting the Pattern to Postgres

The fixture shape is unchanged; only the engine setup differs. Point the engine at a real database, drop StaticPool, and create the schema against a per-run database or schema name:

@pytest_asyncio.fixture(scope="session")
async def engine():
    url = os.environ["TEST_DATABASE_URL"]  # e.g. postgresql+asyncpg://.../app_test
    engine = create_async_engine(url)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()

The db_session and client fixtures need no changes at all — the connection-plus-rollback trick is a SQLAlchemy feature, not a SQLite one. If your project uses migrations, run those instead of create_all so the tests exercise the schema you will actually deploy.

Verification

The pattern is only trustworthy if you can detect it silently breaking, so keep two guards.

Assert that the app is not talking to the production engine:

async def test_override_is_actually_installed(client):
    from myapp import get_session
    assert get_session in app.dependency_overrides

And assert that overrides do not survive the test:

def test_no_overrides_leak_between_tests():
    # Runs after the async tests above; the client fixture must have cleaned up.
    assert app.dependency_overrides == {}

Beyond that, the diagnostic habit is to make one test deliberately commit and confirm a later test cannot see it — which is exactly what the first three tests above encode.

Trade-offs and When Not To

The savepoint trick hides some real behaviour. Because the app's commit() is not a true commit, deferred constraints and commit-time triggers do not fire the way they would in production. For those specific behaviours, write a smaller suite that commits for real and cleans up with truncation.

Session-scoped engines and parallel tests interact badly. Under pytest-xdist each worker is a separate process and needs its own database name, or workers will roll back each other's data. Derive the database name from the worker id.

In-memory SQLite is not your database. It is fast and fine for wiring, validation and transaction-boundary tests. It will not reproduce Postgres locking, JSONB behaviour, dialect-specific SQL or the pool exhaustion covered in Fixing asyncpg Pool Exhaustion. Keep an integration tier on the real engine.

Fixtures that share a session hide concurrency bugs. Every test here runs one request at a time against one session. Code that only breaks under concurrent requests to the same row needs a test that issues concurrent requests — see Concurrent Requests with asyncio.gather.

Use AsyncClient, not TestClient, with these fixtures. The fixtures are async and live on the test's event loop; TestClient runs the app on a different loop of its own. The consequences are laid out in TestClient vs httpx AsyncClient.

FAQ

Why do my async tests fail with async def functions are not natively supported? pytest cannot run a coroutine on its own, so without an async plugin it collects the test, refuses to await it and fails. Install pytest-asyncio and set asyncio_mode to auto, or mark each test with pytest.mark.asyncio.

How do I make each test start from a clean database? Open a connection, begin a transaction on it, bind the session to that connection, and roll the transaction back after the test. Everything the test wrote, including its commits, disappears without dropping or recreating any tables.

How can the test still see rows after the endpoint commits? Bind the session maker to the open connection with join_transaction_mode set to create_savepoint. The endpoint's commit then resolves to releasing a savepoint inside your outer transaction rather than ending it, so the data stays visible and stays undoable.

How do I point the app at the test database? Override the session dependency in app.dependency_overrides with a callable returning the test session, then clear the overrides in the fixture teardown so the override cannot leak into another test.

Why does my session-scoped async fixture raise a different event loop error? A fixture that lives longer than one test must run on a loop that also lives that long. Set asyncio_default_fixture_loop_scope to session and give the tests a matching loop scope, otherwise pytest-asyncio creates a fresh loop per test while the engine is still bound to the first one.

Is in-memory SQLite good enough for testing? It is excellent for fast tests of application logic, wiring and transaction boundaries, but it does not reproduce Postgres dialect features, concurrency or locking. Run the fast suite on SQLite and a smaller integration suite against the real engine.