FastAPI

FastAPI Database Testing with SQLAlchemy and Fixtures

HelpMeTest

HelpMeTest

— 9 min read
FastAPI Database Testing with SQLAlchemy and Fixtures

Database testing is where most FastAPI applications get their testing strategy wrong. The common mistakes: using a shared test database that accumulates state, running migrations on every test, or — worst of all — hitting the production database in CI. This guide shows you how to test FastAPI applications backed by SQLAlchemy correctly, with real isolation, real queries, and real confidence.

The Core Problem: Test Isolation

Every database test needs to answer one question: what state is the database in when this test starts? If the answer is "it depends on what previous tests did," your test suite is broken. Tests will pass locally and fail in CI depending on execution order. You'll spend hours debugging failures that disappear when you run the test in isolation.

The solution is database isolation at the transaction level.

Per-Test Database Transactions with Rollback

The most efficient isolation strategy wraps each test in a transaction and rolls it back when the test finishes. No data ever actually commits, so the database stays clean:

# tests/conftest.py
import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import (
    create_async_engine,
    AsyncSession,
    async_sessionmaker,
)
from sqlalchemy.pool import NullPool
from app.models import Base
from app.dependencies import get_db
from app.main import app

# Use a real test database — not SQLite, which behaves differently from Postgres
DATABASE_URL = "postgresql+asyncpg://postgres:testpassword@localhost:5432/testdb"

@pytest_asyncio.fixture(scope="session")
async def test_engine():
    """Single engine for the entire test session."""
    engine = create_async_engine(
        DATABASE_URL,
        poolclass=NullPool,  # Disable connection pooling for tests
    )
    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 db(test_engine):
    """
    Per-test async session with transaction rollback.
    
    The key: we use a nested transaction (SAVEPOINT) so that even if the
    endpoint code commits, we can roll back the outer transaction.
    """
    async with test_engine.connect() as conn:
        await conn.begin()
        # Start a savepoint that the endpoint code can commit against
        # without actually committing to the database
        async with conn.begin_nested() as savepoint:
            session = AsyncSession(bind=conn, expire_on_commit=False)
            session.sync_session.begin_nested()

            yield session

            await session.close()
        await savepoint.rollback()

@pytest_asyncio.fixture
async def client(db):
    """HTTP client wired to the test session."""
    from httpx import AsyncClient, ASGITransport

    async def override_get_db():
        yield db

    app.dependency_overrides[get_db] = override_get_db

    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://testserver"
    ) as c:
        yield c

    app.dependency_overrides.clear()

The SAVEPOINT trick is crucial. When your endpoint calls session.commit(), it commits the savepoint, not the outer transaction. The rollback at the end undoes everything, including what the endpoint thought it committed.

SQLAlchemy Async Session Fixtures

Different test scenarios need different session configurations. Here are the patterns you'll use most:

# tests/conftest.py (continued)

@pytest_asyncio.fixture
async def fresh_session(test_engine):
    """
    A session that doesn't share state with the app's session.
    Use this for verifying database state after an API call.
    """
    Session = async_sessionmaker(test_engine, expire_on_commit=False)
    async with Session() as session:
        async with session.begin():
            yield session

@pytest_asyncio.fixture
async def read_only_session(test_engine):
    """Read-only session for verifying state without modifying it."""
    async with test_engine.connect() as conn:
        await conn.execution_options(isolation_level="SERIALIZABLE")
        session = AsyncSession(bind=conn)
        yield session
        await session.close()
# tests/test_user_creation.py
from app.models import User, Profile

async def test_create_user_also_creates_profile(client, db):
    response = await client.post("/users", json={
        "email": "alice@example.com",
        "name": "Alice",
        "create_profile": True
    })
    assert response.status_code == 201
    user_id = response.json()["id"]

    # Verify both User and Profile exist in the same transaction
    from sqlalchemy import select
    user = await db.get(User, user_id)
    assert user is not None
    assert user.email == "alice@example.com"

    result = await db.execute(
        select(Profile).where(Profile.user_id == user_id)
    )
    profile = result.scalar_one_or_none()
    assert profile is not None

async def test_user_creation_fails_on_duplicate_email(client, db):
    # Pre-populate
    user = User(email="existing@example.com", name="Existing")
    db.add(user)
    await db.flush()  # flush but not commit — still in our test transaction

    response = await client.post("/users", json={
        "email": "existing@example.com",
        "name": "Duplicate"
    })
    assert response.status_code == 409
    assert "already exists" in response.json()["detail"]

factory_boy with SQLAlchemy

Manually constructing test objects gets tedious. factory_boy lets you define factories that create realistic test data with sensible defaults that you override only when relevant:

pip install factory-boy faker
# tests/factories.py
import factory
import factory.fuzzy
from factory.alchemy import SQLAlchemyModelFactory
from faker import Faker
from app.models import User, Post, Comment, Tag

fake = Faker()

class UserFactory(SQLAlchemyModelFactory):
    class Meta:
        model = User
        sqlalchemy_session_persistence = "flush"

    email = factory.LazyFunction(lambda: fake.unique.email())
    name = factory.LazyFunction(fake.name)
    is_active = True
    is_admin = False
    created_at = factory.LazyFunction(fake.date_time_this_year)

class AdminUserFactory(UserFactory):
    is_admin = True
    email = factory.LazyFunction(lambda: f"admin-{fake.unique.numerify('###')}@example.com")

class PostFactory(SQLAlchemyModelFactory):
    class Meta:
        model = Post
        sqlalchemy_session_persistence = "flush"

    title = factory.LazyFunction(lambda: fake.sentence(nb_words=6))
    content = factory.LazyFunction(lambda: fake.paragraphs(nb=3, ext_word_list=None))
    author = factory.SubFactory(UserFactory)
    status = factory.fuzzy.FuzzyChoice(["draft", "published", "archived"])
    view_count = factory.fuzzy.FuzzyInteger(0, 10000)

class CommentFactory(SQLAlchemyModelFactory):
    class Meta:
        model = Comment
        sqlalchemy_session_persistence = "flush"

    post = factory.SubFactory(PostFactory)
    author = factory.SubFactory(UserFactory)
    content = factory.LazyFunction(lambda: fake.paragraph())
# tests/conftest.py (add this)
@pytest_asyncio.fixture(autouse=True)
async def bind_factories(db):
    """Bind all factories to the test session automatically."""
    from tests.factories import UserFactory, PostFactory, CommentFactory
    for factory_class in [UserFactory, PostFactory, CommentFactory]:
        factory_class._meta.sqlalchemy_session = db
    yield
    for factory_class in [UserFactory, PostFactory, CommentFactory]:
        factory_class._meta.sqlalchemy_session = None
# tests/test_post_endpoints.py
from tests.factories import UserFactory, PostFactory, CommentFactory

async def test_get_posts_returns_only_published(client, db):
    UserFactory.create()  # author
    author = UserFactory.create()

    PostFactory.create_batch(5, status="published", author=author)
    PostFactory.create_batch(3, status="draft", author=author)
    PostFactory.create_batch(2, status="archived", author=author)

    await db.flush()

    response = await client.get("/posts?status=published")
    assert response.status_code == 200
    posts = response.json()["items"]
    assert len(posts) == 5
    assert all(p["status"] == "published" for p in posts)

async def test_post_comment_count_is_accurate(client, db):
    post = PostFactory.create(status="published")
    CommentFactory.create_batch(7, post=post)
    await db.flush()

    response = await client.get(f"/posts/{post.id}")
    assert response.status_code == 200
    assert response.json()["comment_count"] == 7

async def test_admin_can_see_draft_posts(client, db):
    from tests.factories import AdminUserFactory
    admin = AdminUserFactory.create()
    PostFactory.create_batch(3, status="draft")
    await db.flush()

    # Override auth to return admin user
    from app.dependencies import get_current_user
    app.dependency_overrides[get_current_user] = lambda: admin

    response = await client.get("/posts?status=draft")
    assert response.status_code == 200
    assert len(response.json()["items"]) == 3

    from app.main import app
    app.dependency_overrides.clear()

Testing Migrations

Alembic migrations need their own test strategy. You want to verify migrations are reversible and that data survives them:

# tests/test_migrations.py
import pytest
from alembic.config import Config
from alembic import command
from sqlalchemy import create_engine, text, inspect

SYNC_DATABASE_URL = "postgresql://postgres:testpassword@localhost:5432/migration_test_db"

@pytest.fixture(scope="module")
def migration_engine():
    engine = create_engine(SYNC_DATABASE_URL)
    yield engine
    engine.dispose()

@pytest.fixture(scope="module")
def alembic_config():
    cfg = Config("alembic.ini")
    cfg.set_main_option("sqlalchemy.url", SYNC_DATABASE_URL)
    return cfg

def test_migrations_run_clean(migration_engine, alembic_config):
    """Verify all migrations apply without error."""
    # Start from scratch
    with migration_engine.connect() as conn:
        conn.execute(text("DROP SCHEMA public CASCADE"))
        conn.execute(text("CREATE SCHEMA public"))
        conn.commit()

    command.upgrade(alembic_config, "head")

    # Verify final schema matches our models
    inspector = inspect(migration_engine)
    tables = inspector.get_table_names()
    assert "users" in tables
    assert "posts" in tables
    assert "comments" in tables

def test_migrations_are_reversible(migration_engine, alembic_config):
    """Verify we can downgrade all the way back."""
    command.upgrade(alembic_config, "head")
    command.downgrade(alembic_config, "base")

    inspector = inspect(migration_engine)
    tables = inspector.get_table_names()
    # After full downgrade, no app tables should exist
    assert "users" not in tables

def test_data_survives_migration(migration_engine, alembic_config):
    """Upgrade to the version before a migration, insert data, then upgrade."""
    # Upgrade to one revision before the current head
    command.upgrade(alembic_config, "head-1")

    with migration_engine.connect() as conn:
        conn.execute(text(
            "INSERT INTO users (email, name) VALUES ('test@example.com', 'Test User')"
        ))
        conn.commit()

    # Now apply the final migration
    command.upgrade(alembic_config, "head")

    with migration_engine.connect() as conn:
        result = conn.execute(
            text("SELECT email, name FROM users WHERE email = 'test@example.com'")
        )
        row = result.fetchone()

    assert row is not None
    assert row.name == "Test User"

Testing Complex Queries

Repository methods with complex SQLAlchemy queries need unit tests that verify the query logic independently of the HTTP layer:

# app/repositories/post_repository.py
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func, and_, or_, desc
from sqlalchemy.orm import selectinload
from app.models import Post, User, Comment, Tag, post_tags

class PostRepository:
    def __init__(self, session: AsyncSession):
        self.session = session

    async def get_trending(
        self,
        limit: int = 10,
        min_views: int = 100,
        days: int = 7
    ) -> list[Post]:
        from datetime import datetime, timedelta
        since = datetime.utcnow() - timedelta(days=days)

        stmt = (
            select(Post)
            .options(selectinload(Post.author), selectinload(Post.tags))
            .join(Post.author)
            .where(
                and_(
                    Post.status == "published",
                    Post.published_at >= since,
                    Post.view_count >= min_views,
                )
            )
            .order_by(desc(Post.view_count))
            .limit(limit)
        )
        result = await self.session.execute(stmt)
        return result.scalars().unique().all()

    async def search(
        self,
        query: str,
        tags: list[str] | None = None,
        author_id: int | None = None,
    ) -> list[Post]:
        stmt = (
            select(Post)
            .options(selectinload(Post.tags))
            .where(Post.status == "published")
        )

        if query:
            stmt = stmt.where(
                or_(
                    Post.title.ilike(f"%{query}%"),
                    Post.content.ilike(f"%{query}%"),
                )
            )

        if tags:
            stmt = stmt.join(post_tags).join(Tag).where(Tag.name.in_(tags))

        if author_id:
            stmt = stmt.where(Post.author_id == author_id)

        result = await self.session.execute(stmt)
        return result.scalars().unique().all()
# tests/test_post_repository.py
import pytest
import pytest_asyncio
from datetime import datetime, timedelta
from app.repositories.post_repository import PostRepository
from tests.factories import PostFactory, UserFactory, TagFactory

@pytest_asyncio.fixture
async def repo(db):
    return PostRepository(db)

async def test_get_trending_filters_by_min_views(repo, db):
    PostFactory.create_batch(3, status="published", view_count=50)   # below threshold
    PostFactory.create_batch(2, status="published", view_count=500)  # above threshold

    await db.flush()

    results = await repo.get_trending(min_views=100)
    assert len(results) == 2
    assert all(p.view_count >= 100 for p in results)

async def test_get_trending_excludes_old_posts(repo, db):
    old_post = PostFactory.create(
        status="published",
        view_count=9999,
        published_at=datetime.utcnow() - timedelta(days=30)
    )
    recent_post = PostFactory.create(
        status="published",
        view_count=200,
        published_at=datetime.utcnow() - timedelta(hours=1)
    )
    await db.flush()

    results = await repo.get_trending(days=7, min_views=100)
    ids = [p.id for p in results]
    assert old_post.id not in ids
    assert recent_post.id in ids

async def test_search_matches_title_and_content(repo, db):
    PostFactory.create(
        status="published",
        title="Python Testing Guide",
        content="Learn to test your code."
    )
    PostFactory.create(
        status="published",
        title="FastAPI Tutorial",
        content="This guide covers Python web development."
    )
    PostFactory.create(
        status="published",
        title="Unrelated Post",
        content="Nothing to do with testing."
    )
    await db.flush()

    results = await repo.search("Python")
    assert len(results) == 2

async def test_search_by_tag_filters_correctly(repo, db):
    python_tag = TagFactory.create(name="python")
    js_tag = TagFactory.create(name="javascript")

    python_post = PostFactory.create(status="published", tags=[python_tag])
    js_post = PostFactory.create(status="published", tags=[js_tag])
    both_post = PostFactory.create(status="published", tags=[python_tag, js_tag])
    await db.flush()

    results = await repo.search("", tags=["python"])
    ids = {p.id for p in results}
    assert python_post.id in ids
    assert both_post.id in ids
    assert js_post.id not in ids

Isolation Strategies for Different Test Types

Not all tests need the same isolation level. Here's a tiered approach:

# tests/conftest.py — isolation strategies

@pytest_asyncio.fixture(scope="session")
async def static_data(test_engine):
    """
    Seed data that's inserted once and never modified.
    For: read-only reference data, enum values, configuration.
    """
    from app.models import Category
    Session = async_sessionmaker(test_engine)
    async with Session() as session:
        async with session.begin():
            categories = [
                Category(name="Technology", slug="technology"),
                Category(name="Science", slug="science"),
                Category(name="Business", slug="business"),
            ]
            session.add_all(categories)
    yield

@pytest_asyncio.fixture(scope="function")
async def isolated_db(test_engine, static_data):
    """
    Full isolation per test via transaction rollback.
    For: write tests, tests that modify data.
    """
    async with test_engine.connect() as conn:
        transaction = await conn.begin()
        session = AsyncSession(bind=conn, expire_on_commit=False)
        yield session
        await session.close()
        await transaction.rollback()

@pytest.fixture(scope="class")
def shared_test_data(db):
    """
    Shared setup across tests in a class.
    For: tests that read the same pre-populated data.
    Use only when you're certain tests won't modify the data.
    """
    # Populated in conftest or a class-scoped fixture
    pass
# tests/test_category_endpoints.py
class TestCategoryEndpoints:
    """
    These tests only read category data — safe to share across the class.
    """

    async def test_list_categories(self, client, static_data):
        response = await client.get("/categories")
        assert response.status_code == 200
        categories = response.json()
        assert len(categories) == 3
        slugs = {c["slug"] for c in categories}
        assert "technology" in slugs

    async def test_get_category_by_slug(self, client, static_data):
        response = await client.get("/categories/science")
        assert response.status_code == 200
        assert response.json()["name"] == "Science"

    async def test_nonexistent_category_returns_404(self, client):
        response = await client.get("/categories/nonexistent")
        assert response.status_code == 404

Avoiding Common Pitfalls

Don't use SQLite for Postgres applications. SQLite has different behavior around constraints, concurrency, and data types. A test that passes with SQLite may fail with Postgres because SQLite silently ignores some constraint violations.

Don't use expire_on_commit=False globally. It masks bugs where your code accesses attributes after a session is closed. Use it only in test sessions where you genuinely need to access data after commit.

Don't forget await db.flush() after adding factory objects. Without a flush, the data isn't written to the database connection, and your endpoint may not see it.

Always check that your isolation actually works. Write a simple test that adds data, then write a second test that verifies the data isn't there. If the second test fails, your isolation is broken.

# tests/test_isolation_sanity.py
async def test_isolation_part_1(client, db):
    from tests.factories import UserFactory
    UserFactory.create(email="isolation-test@example.com")
    await db.flush()

    response = await client.get("/users?email=isolation-test@example.com")
    assert len(response.json()) == 1  # visible within this test

async def test_isolation_part_2(client, db):
    # This should see 0 users — the previous test's data was rolled back
    from sqlalchemy import select, func
    from app.models import User
    result = await db.execute(select(func.count(User.id)))
    count = result.scalar()
    assert count == 0, f"Isolation failed: {count} users found from previous test"

This sanity check should be the first test you write when setting up database testing. If it fails, fix your isolation before writing any other tests.

Read more

Start now free