A complete technical breakdown of how FastAPI, SQLAlchemy, asyncpg, PostgreSQL, Declarative Base, and Alembic work together as a production system - covering sessions, transactions, rollbacks, relationships, migrations, and architecture.
Most tutorials show you how to get something running. The focus of this article is on how it actually works and why each decision matters when you are building something that must survive real traffic, real failures, and real schema changes over time.
What You Are Actually Building
A backend system is a coordinated pipeline. Every request travels through several layers before a response comes back.
Client → API Layer → Business Logic → Database Layer → Storage Engine → Response
Each tool in the stack owns one layer of this pipeline:
| Tool | Responsibility |
|---|---|
| FastAPI | Handles HTTP requests and routing |
| SQLAlchemy | Translates Python objects to SQL |
| asyncpg | Async driver for PostgreSQL communication |
| PostgreSQL | Persistent storage |
| Declarative Base | Defines schema as Python classes |
| Alembic | Manages schema changes over time |
The reason most backend bugs are hard to fix is not incorrect code. It is incorrect mental models of what each layer is doing.
Sync vs Async: Why the Model Matters
Your system can handle requests in two ways: blocking one at a time, or handling many concurrently.
Synchronous means each request holds a worker thread until the database responds:
def get_users(db):
return db.query(User).all()
Asynchronous means the event loop handles other work while waiting for I/O:
async def get_users(db):
result = await db.execute(select(User))
return result.scalars().all()
Key insight: async does not make individual queries faster. It increases throughput by allowing hundreds of concurrent requests without spinning up hundreds of threads.
FastAPI runs on Python's asyncio event loop. Everything else in this stack is chosen to stay compatible with that.
Declarative Base: Defining Schema in Python
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, Integer, String
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
name = Column(String)
email = Column(String)
This class is three things at once: a Python object, a table definition, and a mapping layer. Alembic reads this to generate migrations.
Async Engine and asyncpg
from sqlalchemy.ext.asyncio import create_async_engine
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/db"
engine = create_async_engine(DATABASE_URL)
Common mistake: using postgresql:// instead of postgresql+asyncpg://. It will not error immediately, but it will block your event loop under load.
Sessions: The Most Critical Concept
A session is not a database connection. It is a unit of work manager. It tracks modified objects, manages the transaction lifecycle, and handles cleanup.
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import sessionmaker
SessionLocal = sessionmaker(
bind=engine,
class_=AsyncSession,
expire_on_commit=False
)
expire_on_commit=False keeps object data accessible after commit. Without it, accessing attributes after commit in an async context can raise errors because the session may already be closed.
Dependency Injection: Session Management at Scale
from typing import AsyncGenerator
async def get_db() -> AsyncGenerator[AsyncSession, None]:
async with SessionLocal() as session:
yield session
FastAPI calls get_db(), creates a session, injects it into your route, and closes it after the response. Even on exceptions.
@app.get("/users")
async def get_users(db: AsyncSession = Depends(get_db)):
result = await db.execute(select(User))
return result.scalars().all()
Without this pattern, connection leaks are almost inevitable as the codebase grows.
Transactions and Rollbacks
Every write operation in production must be treated as potentially failing.
try:
db.add(user)
db.add(audit_log)
await db.commit()
except Exception:
await db.rollback()
raise
Without rollback: partial writes, inconsistent state, very hard to debug.
With rollback: database stays in its previous clean state.
Relationships: Foreign Keys and Loading Strategies
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
name = Column(String)
posts = relationship("Post", back_populates="user")
class Post(Base):
__tablename__ = "posts"
id = Column(Integer, primary_key=True)
title = Column(String)
user_id = Column(Integer, ForeignKey("users.id"))
user = relationship("User", back_populates="posts")
Loading relationships efficiently matters at scale. Use selectinload to avoid the N+1 problem:
result = await db.execute(
select(User).options(selectinload(User.posts))
)
selectinload uses a second SELECT. joinedload uses a JOIN. Understanding the difference affects query performance directly.
Alembic: Schema Evolution
# Generate migration after changing a model
alembic revision --autogenerate -m "add email to users"
# Apply it
alembic upgrade head
# Reverse it
alembic downgrade -1
Critical setup: Alembic must import all your models in env.py or it will not detect their changes.
# alembic/env.py
from app.db.base import Base
from app.db.models import user, post
target_metadata = Base.metadata
Production Project Structure
app/
├── main.py
├── api/
│ └── v1/
│ └── routes/
├── core/
│ └── config.py
├── db/
│ ├── base.py
│ ├── session.py
│ └── models/
├── schemas/
├── services/
└── repositories/
Routes stay thin. Services hold business logic. Repositories handle data access. This separation keeps large codebases testable and maintainable.
Common Mistakes That Break Production Systems
# Wrong: sync query in async context
session.query(User).all()
# Wrong: missing await
db.execute(select(User))
# Wrong: manual session (no lifecycle guarantee)
db = SessionLocal()
# Wrong: sync URL with async engine
DATABASE_URL = "postgresql://..."
# Wrong: no rollback on failure
db.add(user)
await db.commit()
Every one of these mistakes has a production impact. None of them cause immediate errors in development.
Conclusion
A modern Python backend is a layered system where each component has a specific responsibility. The most important things to get right:
- Use async throughout. One sync database call can block your event loop and degrade performance for all concurrent requests.
- Use dependency injection for sessions. Manual session management leads to leaks.
- Always handle rollbacks on writes. Partial state is worse than no state.
- Use Alembic from day one. Manually altering production databases is how data gets corrupted.
- Structure your project into layers. Routes, services, repositories, and schemas each have a job.
Understanding why this stack is designed the way it is separates engineers who can debug it from engineers who can only copy it.
I build real-world data systems and write about what I learn. Follow for more backend engineering and data content. GitHub: https://github.com/Rozieroz
Top comments (0)