Test FastAPI endpoints against a real database

Approach

Run PostgreSQL in Docker/Testcontainers, apply Alembic migrations once per session, and for each test open a connection with an outer transaction, bind a session to it with join_transaction_mode="create_savepoint", override get_db, and roll back afterwards.

Solution

# tests/conftest.py
import pytest
from alembic import command
from alembic.config import Config
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine

TEST_URL = "postgresql+asyncpg://test:test@localhost:5433/test"


@pytest.fixture(scope="session", autouse=True)
def migrate():
    cfg = Config("alembic.ini")
    cfg.set_main_option("sqlalchemy.url", TEST_URL)
    command.upgrade(cfg, "head")


@pytest.fixture
async def db():
    engine = create_async_engine(TEST_URL)
    async with engine.connect() as conn:
        outer = await conn.begin()
        session = AsyncSession(bind=conn, join_transaction_mode="create_savepoint", expire_on_commit=False)
        app.dependency_overrides[get_db] = lambda: session
        yield session
        await session.close()
        await outer.rollback()
    app.dependency_overrides.clear()
    await engine.dispose()


@pytest.fixture
async def client(db):
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
        yield c


@pytest.mark.anyio
async def test_create_order(client, db):
    resp = await client.post("/orders", json={"items": [{"sku": "A1", "qty": 2}]})
    assert resp.status_code == 201
    assert await db.scalar(select(func.count()).select_from(Order)) == 1

Pitfalls

• SQLite instead of PostgreSQL.
• Forgetting to clear dependency_overrides.
• Event loop mismatches between async fixtures and the client (use one test runner config).

Variations

• Truncate tables between tests if code under test uses multiple sessions/connections.

Further reading: SQLAlchemy: Session in an external transaction