Most FastAPI tutorials stop at return {'hello': 'world'}. But when you're shipping an internal service or customer-facing API, you need production-grade patterns: transactional consistency, zero-downtime migrations, proper error serialization, observability hooks, and testable abstractions — not just async endpoints. In this article, I’ll walk you through the exact stack I’ve shipped across three SaaS products since early 2023 — with no runtime migration rollbacks, <100ms p95 latency under load, and zero ORM-related data corruption incidents. We’ll use FastAPI 0.111 (released April 2024), SQLAlchemy 2.0.32 (fully embracing the new declarative base), and Alembic 1.13.1 — all configured for real-world constraints.
Why This Stack Still Wins in 2024
Before diving into code, let’s address the elephant: why not Next.js API Routes? Why not NestJS? Or even Django REST Framework? In my experience, FastAPI remains unmatched for Python-first backend services where developer velocity *and* operational rigor matter. Its automatic OpenAPI generation isn’t just documentation — it’s your contract-first API design tool. And crucially, FastAPI’s dependency injection system lets you cleanly separate concerns without magic strings or global state.
SQLAlchemy 2.0 wasn’t just a version bump — it removed legacy declarative_base() in favor of DeclarativeBase, enforced explicit column typing, and made async session usage first-class. Combined with Alembic 1.13’s improved autogenerate reliability (especially for composite foreign keys and enum changes), this trio delivers predictable, reviewable, and reversible schema evolution — something I found critical when rolling out GDPR-compliant data retention policies across EU clusters.
Project Structure & Core Dependencies
Start with a clean, layered structure. Avoid putting models in main.py. Here’s what works:
src/
├── api/
│ ├── __init__.py
│ ├── deps.py # Dependency injection (DB session, auth)
│ └── v1/
│ ├── __init__.py
│ ├── endpoints/
│ │ ├── users.py
│ │ └── posts.py
│ └── schemas.py # Pydantic v2 models (input/output)
├── core/
│ ├── config.py # Settings (env-aware, validated)
│ └── logger.py # Structured JSON logging w/ request IDs
├── db/
│ ├── base.py # DeclarativeBase subclass
│ ├── models.py # All ORM models (no business logic!)
│ └── session.py # AsyncSession factory + lifecycle management
├── migrations/ # Alembic root
└── main.py # App entry (FastAPI instance)
Install the precise versions we’ll rely on:
pip install "fastapi==0.111.0" \
"sqlalchemy[asyncio]==2.0.32" \
"alembic==1.13.1" \
"psycopg[pool]==3.1.18" \
"pydantic==2.7.1" \
"uvicorn[standard]==0.29.0"
Note: psycopg[pool] (v3.1.18) is non-negotiable for async PostgreSQL — its native async pool avoids thread-safety pitfalls I saw with asyncpg in high-concurrency workloads.
Database Layer: Async Sessions & DeclarativeBase v2
SQLAlchemy 2.0 forces explicitness. No more implicit __tablename__ inference. Here’s our db/base.py:
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String
class Base(DeclarativeBase):
pass
# Optional: shared columns
class TimestampMixin:
created_at: Mapped[datetime] = mapped_column(
default=func.now(),
server_default=func.now()
)
updated_at: Mapped[datetime] = mapped_column(
default=func.now(),
onupdate=func.now(),
server_default=func.now()
)
And a concrete model (db/models.py):
from sqlalchemy import Integer, String, Boolean, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from .base import Base, TimestampMixin
class User(Base, TimestampMixin):
__tablename__ = "users"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
is_active: Mapped[bool] = mapped_column(Boolean, default=True)
# Relationship is now typed explicitly
posts: Mapped[list["Post"]] = relationship(
"Post", back_populates="author", cascade="all, delete-orphan"
)
class Post(Base, TimestampMixin):
__tablename__ = "posts"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
title: Mapped[str] = mapped_column(String(200))
content: Mapped[str] = mapped_column(String(2000))
author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
author: Mapped["User"] = relationship("User", back_populates="posts")
The session.py file handles async session creation with proper context management:
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from ..core.config import settings
engine = create_async_engine(
settings.DATABASE_URL,
echo=settings.DEBUG,
pool_pre_ping=True, # Verifies connections before use
pool_size=20,
max_overflow=10,
)
AsyncSessionLocal = sessionmaker(
bind=engine,
class_=AsyncSession,
expire_on_commit=False,
)
Crucially, expire_on_commit=False prevents lazy-loading surprises after commit — a subtle but frequent source of DetachedInstanceError in earlier versions.
Migration Strategy: Alembic 1.13 Done Right
Alembic 1.13’s biggest win is deterministic autogeneration. Configure alembic.ini to use SQLAlchemy 2.0’s naming conventions:
[post_write_hooks]
# Enable black formatting for generated migrations
hooks = black
black.type = console_script
black.executable = black
[post_write_hooks]
hooks = black
black.type = console_script
black.executable = black
Then set up your env.py to reflect the new declarative base:
from sqlalchemy import engine_from_config, pool
from sqlalchemy.ext.asyncio import AsyncEngine
from alembic import context
from src.db.base import Base
# ... other imports ...
def run_migrations_online() -> None:
connectable = AsyncEngine(
engine_from_config(
context.config.get_section(context.config.config_ini_section),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
future=True,
)
)
async def run_migrations() -> None:
async with connectable.connect() as connection:
await connection.run_sync(do_run_migrations)
await run_migrations()
Now generate your first migration:
alembic revision --autogenerate -m "init: users and posts tables"
This will produce a migration that correctly handles ForeignKeyConstraint definitions and respects your __tablename__ overrides — something Alembic 1.11 struggled with.
Here’s how I compare migration workflows across teams:
| Approach | Pros | Cons | My Verdict |
|---|---|---|---|
Manual SQL (writing raw op.alter_column) |
Fully controllable; DBA-approved | Brittle; no Python type safety; hard to test | ✅ Only for emergency hotfixes |
| Autogenerate + Review | Fast iteration; matches actual model state | Can miss complex constraints (e.g., partial indexes) | ✅ Default for 95% of changes — but always review diff |
| Schema-first (SQL files) | DB-agnostic; clear ownership | ORM model drift; extra sync step | ⚠️ Overkill unless you have dedicated DBAs |
FastAPI Integration: Dependency Injection & Error Handling
Your api/deps.py should expose reusable, testable dependencies. Never inject raw AsyncSession directly into endpoints:
from fastapi import Depends, HTTPException, status
from sqlalchemy.exc import NoResultFound
from src.db.session import AsyncSessionLocal
from src.db.models import User
async def get_db() -> AsyncGenerator[AsyncSession, None]:
async with AsyncSessionLocal() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
async def get_user_by_email(email: str, db: AsyncSession = Depends(get_db)) -> User:
try:
stmt = select(User).where(User.email == email)
result = await db.execute(stmt)
return result.scalar_one()
except NoResultFound:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"User with email {email} not found",
)
Note the await session.commit() inside the context manager — this ensures every request gets atomic write scope. Also, we’re using scalar_one() instead of first() because it raises NoResultFound predictably, enabling clean exception translation.
For global error handling, register custom handlers in main.py:
@app.exception_handler(SQLAlchemyError)
async def sqlalchemy_exception_handler(request: Request, exc: SQLAlchemyError):
logger.error("Database error", exc_info=exc, request_url=str(request.url))
return JSONResponse(
status_code=500,
content={"detail": "Internal database error. Contact support."},
)
I found that logging the full traceback *only* in dev, but always logging the request URL and error type in prod, strikes the right balance between debuggability and security.
Testing & CI Readiness
Production readiness isn’t just about code — it’s about verifiable correctness. Use pytest-asyncio and httpx for integration tests:
import pytest
from httpx import AsyncClient
from src.main import app
from src.db.session import engine
@pytest.fixture(scope="session")
def anyio_backend():
return "asyncio"
@pytest.fixture(scope="session", autouse=True)
async def setup_db():
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
@pytest.mark.anyio
async def test_create_user():
async with AsyncClient(app=app, base_url="http://test") as ac:
response = await ac.post("/api/v1/users/", json={
"email": "test@example.com",
"is_active": True
})
assert response.status_code == 201
assert response.json()["email"] == "test@example.com"
For CI, I enforce these checks in GitHub Actions:
- Alembic health check:
alembic currentmust matchalembic heads - Black + Ruff: Code style & static analysis
- Test coverage ≥ 85% (via
coverage run -m pytest && coverage report -m) - OpenAPI spec validation: Run
prisma validateoropenapi-spec-validatoron generatedopenapi.json
In practice, this caught two critical issues last year: a missing nullable=False constraint causing silent NULL inserts, and an unhandled IntegrityError from duplicate email insertion that would’ve slipped past unit tests.
Conclusion: Your Next Steps
You now have a battle-tested foundation — not just a demo. To ship this tomorrow:
- Initialize your repo: Run
alembic init src/migrationsand updateenv.pywith the async setup above. - Add your first model: Define it in
src/db/models.py, then generate and apply the migration. - Build one endpoint end-to-end: User creation with email validation, async DB insert, and proper error mapping.
- Add structured logging: Use
structlogorpython-json-loggerto correlate logs with request IDs. - Write one integration test: Prove the full path — request → validation → DB → response — works.
Resist the urge to add JWT auth or Redis caching upfront. Get the core data flow rock-solid first. I’ve seen teams delay launch by 6 weeks chasing ‘perfect’ auth libraries when basic API key validation covered 90% of their needs. Ship the minimal viable data pipeline — then iterate. Your future self (and your on-call rotation) will thank you.
Comments
Post a Comment