Testing FastAPI + SQLAlchemy with Real PostgreSQL Fixtures: No More Mocking Misery
I spent three months debugging why a query worked in tests but failed in production. The culprit? I'd mocked the entire database layer.
Mocks are seductive. They're fast, isolated, and you control everything. But they're also liars. They hide migration bugs, concurrency issues, transaction edge cases, and the thousand small ways your ORM behaves differently under real database constraints. I'd rather catch those bugs in CI than at 2 AM in production.
This post shows you how to test FastAPI + SQLAlchemy against real PostgreSQL instances—cheaply, quickly, and with perfect isolation. No mocking the data layer. No false confidence.
Why Real Databases Beat Mocks
Before diving into code, let me be direct about why I changed my mind.
Mocks hide real problems. When you mock session.query(), you're not testing how SQLAlchemy constructs queries, handles N+1 problems, or manages transactions. You're testing your test setup.
Migrations are invisible to mocks. I once added a NOT NULL constraint in a migration. Tests passed (mocked). Production broke (real database). Now I run real migrations in every test database.
Concurrency and locks don't exist in mocks. Row-level locking, deadlock conditions, and transaction isolation levels are real problems in multi-tenant systems. Mocks can't catch them.
Your ORM behaves differently than you expect. SQLAlchemy's relationship loading, cascade deletes, and lazy vs. eager loading have edge cases. Real tests find them.
The trade-off isn't as bad as it sounds. Docker + pytest fixtures make spinning up isolated PostgreSQL databases fast—we're talking milliseconds for setup, and modern CI can parallelize tests across multiple database instances.
The Setup: Docker Compose + Pytest Fixtures
I use a minimal Docker Compose setup that spins up PostgreSQL and tears it down cleanly between test runs.
docker-compose.test.yml:
version: '3.8'
services:
postgres-test:
image: postgres:16-alpine
environment:
POSTGRES_USER: test_user
POSTGRES_PASSWORD: test_password
POSTGRES_DB: test_db
ports:
- "5433:5432"
healthcheck:
test: ["CMD-SHELL", "pg_isready -U test_user"]
interval: 1s
timeout: 5s
retries: 10
tmpfs:
- /var/lib/postgresql/data
The tmpfs line is crucial—it stores the database in RAM, making writes blazingly fast. Perfect for tests.
conftest.py — this is where the magic happens:
import pytest
import subprocess
import time
from sqlalchemy import create_engine, text
from sqlalchemy.pool import NullPool
from sqlalchemy.orm import sessionmaker, Session
# Import your models
from app.models import Base
from app.database import get_db
@pytest.fixture(scope="session")
def docker_compose():
"""Spin up Docker Compose once per test session."""
subprocess.run(
["docker-compose", "-f", "docker-compose.test.yml", "up", "-d"],
check=True,
cwd=".",
)
# Wait for postgres to be ready
max_retries = 30
for i in range(max_retries):
try:
engine = create_engine(
"postgresql://test_user:test_password@localhost:5433/test_db"
)
with engine.connect() as conn:
conn.execute(text("SELECT 1"))
break
except Exception as e:
if i == max_retries - 1:
raise
time.sleep(0.5)
yield
# Tear down
subprocess.run(
["docker-compose", "-f", "docker-compose.test.yml", "down"],
check=True,
cwd=".",
)
@pytest.fixture
def db_session(docker_compose):
"""Create a fresh database session for each test."""
# Create engine with NullPool—don't reuse connections between tests
engine = create_engine(
"postgresql://test_user:test_password@localhost:5433/test_db",
poolclass=NullPool,
)
# Run migrations (I use Alembic; you could also use Base.metadata.create_all)
subprocess.run(
["alembic", "upgrade", "head"],
check=True,
cwd=".",
)
SessionLocal = sessionmaker(bind=engine, expire_on_commit=False)
session = SessionLocal()
yield session
# Cleanup: rollback any uncommitted transactions
session.rollback()
session.close()
# Drop all tables between tests for true isolation
Base.metadata.drop_all(engine)
@pytest.fixture
def client(db_session):
"""Provide a FastAPI test client with dependency injection."""
from fastapi.testclient import TestClient
from app.main import app
def override_get_db():
yield db_session
app.dependency_overrides[get_db] = override_get_db
return TestClient(app)
Notice the key decisions:
NullPool: Each test gets a fresh connection. No connection reuse surprises.
drop_all()after each test: Complete isolation. No shared state.- Alembic migrations run in every test: You catch migration bugs immediately.
Real Test Example: Catching What Mocks Miss
Let's test a user creation endpoint with a unique email constraint:
from app.models import User
from datetime import datetime
def test_create_user_success(client, db_session):
"""Test successful user creation."""
response = client.post(
"/users",
json={"email": "[email protected]", "name": "Alice"},
)
assert response.status_code == 201
assert response.json()["email"] == "[email protected]"
# Verify it's actually in the database
user = db_session.query(User).filter_by(email="[email protected]").first()
assert user is not None
assert user.name == "Alice"
def test_create_duplicate_email_fails(client, db_session):
"""Test that duplicate emails are rejected.
This catches:
- Whether your database constraint actually exists
- Whether your error handling returns the right status code
- Whether the transaction rolled back properly
"""
# Insert first user
client.post(
"/users",
json={"email": "[email protected]", "name": "Bob"},
)
# Try to insert duplicate
response = client.post(
"/users",
json={"email": "[email protected]", "name": "Bob 2"},
)
assert response.status_code == 409 # Conflict
assert "already exists" in response.json()["detail"].lower()
# Verify only one user exists
count = db_session.query(User).filter_by(email="[email protected]").count()
assert count == 1
def test_n_plus_one_query_detection(client, db_session):
"""Catch N+1 queries against real PostgreSQL."""
from sqlalchemy import event
# Track queries
queries = []
@event.listens_for(engine, "before_cursor_execute")
def receive_before_cursor_execute(conn, cursor, statement, params, context, executemany):
queries.append(statement)
# Create users with posts
for i in range(5):
user = User(email=f"user{i}@example.com", name=f"User {i}")
db_session.add(user)
db_session.commit()
# Fetch users and their posts efficiently
users = db_session.query(User).options(joinedload(User.posts)).all()
# Against a real database, you can verify query patterns
# Mocks can't catch these subtle performance bugs
assert len(queries) < 10 # Rough check; adjust for your schema
That last test is impossible to write meaningfully against mocks. With real PostgreSQL, you actually catch N+1 queries, missing indexes, and transaction issues.
Gotcha: Test Database Concurrency
Here's what burned me: I created a test for concurrent writes, but because each test got its own db_session, I wasn't actually testing multiple concurrent connections.
Solution: When you need true concurrency tests, use separate database sessions:
def test_concurrent_writes(db_session):
"""Test actual concurrent database writes."""
from concurrent.futures import ThreadPoolExecutor
from app.database import SessionLocal
def create_user(email: str):
session = SessionLocal()
try:
user = User(email=email, name=email.split("@")[0])
session.add(user)
session.commit()
return user.id
finally:
session.close()
# Run 10 writes concurrently
with ThreadPoolExecutor(max_workers=10) as executor:
futures = [
executor.submit(create_user, f"user{i}@example.com")
for i in range(10)
]
results = [f.result() for f in futures]
# Verify all succeeded
assert len(set(results)) == 10 # All IDs unique
This actually stress-tests your database and connection pool. Mocks wouldn't catch connection pool exhaustion or deadlocks.
Speed: It's Faster Than You Think
People worry real database tests are slow. They're not—especially with tmpfs and good fixture design:
- Database spin-up: ~2s (once per session)
- Per-test setup: ~50ms
- Typical test execution: ~100ms
A 100-test suite runs in under 15 seconds locally. In CI with parallelization
SOCIAL SHARE CARD GENERATOR