Overriding the database session dependency in FastAPI tests

Override the session dependency with a factory bound to a connection whose transaction the test rolls back — async_sessionmaker(bind=connection, join_transaction_mode="create_savepoint") — so the endpoint's own commit() releases a savepoint instead of persisting, and teardown is one rollback. This guide belongs to integrating SQLAlchemy async with FastAPI and Starlette.

Quick Answer

Without an override, the endpoint opens its own session on its own connection, so the test and the endpoint cannot see each other's data.

Before — the test and the endpoint use different transactions:

import httpx
import pytest

from shop.main import app


@pytest.mark.asyncio
async def test_create_order(session, customer_factory):
    customer_id = await customer_factory()
    await session.commit()          # the test's transaction

    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.post("/orders", json={"customer_id": customer_id})

    assert response.status_code == 201
# The endpoint used the application engine — possibly a different database entirely —
# and anything it wrote is invisible to the test session.

After — one connection, one transaction, rolled back at teardown:

from collections.abc import AsyncIterator

import httpx
import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from shop.db import get_session
from shop.main import app


@pytest_asyncio.fixture
async def db_session(test_engine) -> AsyncIterator[AsyncSession]:
    async with test_engine.connect() as connection:
        await connection.begin()                     # the outer transaction
        factory = async_sessionmaker(
            bind=connection,
            expire_on_commit=False,
            join_transaction_mode="create_savepoint",
        )
        async with factory() as session:
            yield session
        await connection.rollback()                  # nothing persists


@pytest_asyncio.fixture
async def client(db_session) -> AsyncIterator[httpx.AsyncClient]:
    async def override() -> AsyncIterator[AsyncSession]:
        yield db_session

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

join_transaction_mode="create_savepoint" is the key setting: when the endpoint calls commit(), SQLAlchemy releases a savepoint inside the test's transaction rather than committing it, so the write is visible to the test and still disappears at rollback.

Execution Context & Async Workflow Integration

FastAPI resolves dependencies by callable identity, and app.dependency_overrides is a dictionary keyed on the original callable. Replacing get_session therefore changes what every endpoint receives without touching application code — which is the whole point: the code under test is the production code.

The transaction arrangement is what makes the tests both isolated and fast. The fixture holds one connection with an outer transaction. A session factory bound to that connection — not to the engine — means every session created from it joins the existing transaction rather than starting its own. join_transaction_mode="create_savepoint" tells SQLAlchemy what to do when such a session commits: begin a savepoint on entry and release it on commit, so the application's commit() succeeds, its writes become visible, and the outer transaction still owns everything. One rollback() at teardown undoes the lot in microseconds.

That is the same fixture described in rolling back database state between async tests; the addition here is the override that makes an endpoint use it.

httpx.ASGITransport calls the application in-process, with no socket and no server. That keeps tests fast and means the application's lifespan does not run unless asked — which is usually what you want in a test, because the lifespan would create the real engine. When a test does need the lifespan (to check start-up behaviour), httpx.ASGITransport combined with LifespanManager from asgi-lifespan runs it explicitly.

Two async-specific details matter.

The event loop must be shared between the fixture and the request. pytest-asyncio's default is a loop per test function, which is fine as long as the engine fixture is scoped the same way or the loop scope is widened; mixing a session-scoped engine with function-scoped loops produces got Future attached to a different loop, the failure described in disposing async engines on shutdown and in forked workers.

Background tasks run before the client call returns with ASGITransport, so their effects are assertable immediately. But a task that opens its own session from the application factory — which is what it should do — will not see the test's uncommitted transaction. Either override that factory too, or accept that such tests need committed fixture data and truncation-based cleanup, as running background tasks with a fresh AsyncSession in FastAPI notes.

Resolving Warnings, Errors & Common Mistakes

The persistence case is the one worth verifying deliberately, because a suite that accidentally commits passes for a long time before the leakage causes a confusing failure. One test proves the arrangement works:

import pytest
from sqlalchemy import func, select

from shop.models import Order


@pytest.mark.asyncio
async def test_isolation_holds(client, db_session, customer_factory):
    customer_id = await customer_factory()
    response = await client.post("/orders", json={"customer_id": customer_id})
    assert response.status_code == 201
    # Visible inside the test's transaction:
    assert await db_session.scalar(select(func.count(Order.id))) == 1


@pytest.mark.asyncio
async def test_previous_test_left_nothing(db_session):
    assert await db_session.scalar(select(func.count(Order.id))) == 0

Run in order, those two assert both halves: the endpoint's write is visible during the test and gone afterwards. If the second fails, the savepoint arrangement is not in place and the suite is accumulating state.

Overriding more than the session is sometimes necessary and worth keeping deliberate. Authentication is the common one — a current_user dependency replaced with a fixed user — and it is better to override the dependency than to construct tokens, because the test then exercises the endpoint rather than the auth provider:

from shop.auth import current_user
from shop.models import User


@pytest_asyncio.fixture
async def as_admin(client, db_session):
    admin = User(email="admin@example.com", is_staff=True)
    db_session.add(admin)
    await db_session.flush()
    app.dependency_overrides[current_user] = lambda: admin
    yield client
    app.dependency_overrides.pop(current_user, None)

Advanced: What to Assert, and What Not to Mock

An endpoint test is expensive compared with a unit test, so it should assert the things only it can. Four are worth the cost.

The committed state, read from the database. A response body reflects what the handler intended; the database reflects what happened. Asserting both catches serialisation bugs and transaction bugs separately:

import pytest
from sqlalchemy import select

from shop.models import Order


@pytest.mark.asyncio
async def test_create_order_persists_the_total(client, db_session, customer_factory):
    customer_id = await customer_factory()

    response = await client.post(
        "/orders",
        json={"customer_id": customer_id, "lines": [{"sku": "A1", "quantity": 2}]},
    )

    assert response.status_code == 201
    body = response.json()

    order = await db_session.scalar(select(Order).where(Order.id == body["id"]))
    assert order is not None
    assert order.total_cents == body["total_cents"]     # response matches the row
    assert order.status == "pending"

The contract on failure. Validation errors, missing resources and conflicts have status codes and shapes that clients depend on. A test per failure mode is cheap and prevents a refactor from turning a 409 into a 500.

Transaction boundaries. An endpoint that should write nothing on failure is worth asserting: trigger the failure, then count the rows. This catches a handler that commits before validating.

The database, not a mock of it. Mocking the session in an endpoint test removes the only thing the test was for. The interactions that break in production — constraint violations, transaction boundaries, lazy loads that raise under async, the actual SQL — are exactly what a mock cannot reproduce. Mock the things outside the system: the payment provider, the mail sender, the object store.

@pytest.fixture
def payments(monkeypatch):
    captured = []

    async def capture(order_id: int, amount_cents: int) -> str:
        captured.append((order_id, amount_cents))
        return "ref_123"

    monkeypatch.setattr("shop.payments.capture", capture)
    return captured

That distinction — real database, fake outside world — is what makes endpoint tests worth their runtime. A suite that mocks the database runs faster and tells you only that the Python is syntactically connected.

For the handful of tests that genuinely need committed data — background tasks, anything reading through a second connection — keep a separate fixture that commits and cleans up by truncation, and mark those tests so the difference is visible. Two isolation strategies in one suite is fine as long as each test says which it uses.

Structuring the Fixture Stack

The fixtures in this guide compose into a small stack, and getting the order and the scopes right once makes every endpoint test afterwards a few lines.

# tests/conftest.py
from collections.abc import AsyncIterator

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

from shop.db import get_session
from shop.main import app
from shop.models import Base


@pytest_asyncio.fixture(scope="session")
async def test_engine(postgres_url: str):
    """One engine and one schema for the whole run."""
    engine = create_async_engine(postgres_url)
    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(test_engine) -> AsyncIterator[AsyncSession]:
    """A session inside a transaction that is rolled back after the test."""
    async with test_engine.connect() as connection:
        await connection.begin()
        factory = async_sessionmaker(
            bind=connection,
            expire_on_commit=False,
            join_transaction_mode="create_savepoint",
        )
        async with factory() as session:
            yield session
        await connection.rollback()


@pytest_asyncio.fixture
async def client(db_session) -> AsyncIterator[httpx.AsyncClient]:
    async def override() -> AsyncIterator[AsyncSession]:
        yield db_session

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

Four decisions are encoded there. The engine and schema are session-scoped, so create_all runs once rather than per test. The transaction is function-scoped, which is where the isolation comes from. The override is set and cleared alongside the client, so no test can leak it. And the event loop scope must match the engine scope — set asyncio_default_fixture_loop_scope = "session" in the pytest configuration, or the session-scoped engine will be used from the wrong loop.

Building the schema with create_all is a deliberate simplification worth noting: it creates what the models describe, not what the migrations produce. For a suite that also needs to verify the migration path, a second session-scoped fixture that runs alembic upgrade head instead is the alternative — the arrangement in running Alembic migrations programmatically from async code.

The postgres_url fixture is whatever provides a throwaway database: a testcontainer, a CI service, or a locally running instance with a dedicated database name. Using a real PostgreSQL rather than SQLite is what makes these tests meaningful, for the reasons in enabling SQLite foreign keys and WAL mode with aiosqlite.

Frequently Asked Questions

Related

• Integrating SQLAlchemy Async with FastAPI and Starlette — The parent guide: request-scoped sessions and dependencies.
• Rolling back database state between async tests — The transaction fixture this builds on.
• Using pytest-asyncio fixtures with AsyncSession — Event loop scopes and fixture ordering.
• Running background tasks with a fresh AsyncSession in FastAPI — Why some tests need committed data.