Database Architecture: SQLAlchemy, Migrations & Performance

ORMs are powerful but dangerous without understanding what happens underneath. This chapter covers production database patterns: async sessions, migration strategies, connection pooling, and the performance patterns that keep your API fast at scale.

SQLAlchemy 2.0 — The Modern Way

SQLAlchemy 2.0 (2023) unified the Core and ORM APIs with a consistent select-based approach:

# ❌ Old SQLAlchemy 1.x style (deprecated)
users = session.query(User).filter(User.age > 18).all()

# ✅ SQLAlchemy 2.0 style
from sqlalchemy import select, update, delete

users = session.execute(
    select(User).where(User.age > 18)
).scalars().all()

Async Database Architecture

FastAPI's async nature demands async database access. Synchronous ORM calls block the event loop:

SQLModel — The Best of Both Worlds

SQLModel combines SQLAlchemy ORM with Pydantic validation in a single model class:

from sqlmodel import SQLModel, Field, Relationship, Session, select

class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    username: str = Field(index=True, unique=True)
    email: str
    is_active: bool = True
    
    # Relationship
    posts: list["Post"] = Relationship(back_populates="author")

class Post(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    title: str
    content: str
    user_id: int = Field(foreign_key="user.id")
    
    author: User = Relationship(back_populates="posts")

# Same model works for API AND database!
@app.post("/users/", response_model=UserRead)
def create_user(user: UserCreate, session: Session = Depends(get_session)):
    db_user = User.model_validate(user)
    session.add(db_user)
    session.commit()
    session.refresh(db_user)
    return db_user

Alembic Migrations — Version Your Schema

Never modify your database schema manually. Alembic tracks every change as a migration file:

# Install
pip install alembic

# Initialize
alembic init alembic

# Generate migration from model changes
alembic revision --autogenerate -m "Add users table"

# Apply migration
alembic upgrade head

# Rollback one version
alembic downgrade -1

Migration file example:

# alembic/versions/abc123_add_users.py
def upgrade():
    op.create_table(
        'users',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('username', sa.String(50), unique=True, nullable=False),
        sa.Column('email', sa.String(100), nullable=False),
        sa.Column('is_active', sa.Boolean(), default=True),
        sa.Column('created_at', sa.DateTime(), server_default=sa.func.now()),
    )
    op.create_index('ix_users_username', 'users', ['username'])

def downgrade():
    op.drop_table('users')

Query Optimization Patterns

Connection Pool Configuration

For production with SQLAlchemy:

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

engine = create_async_engine(
    DATABASE_URL,
    pool_size=20,              # Base pool connections
    max_overflow=10,            # Extra connections under load
    pool_pre_ping=True,         # Verify connection before use
    pool_recycle=3600,          # Recycle connections hourly
    echo=False,                 # Set True for debugging
    connect_args={
        "statement_cache_size": 0,   # Required for PostgreSQL
        "prepared_statement_cache_size": 0,
    }
)

# Session factory — create per request
AsyncSessionLocal = async_sessionmaker(
    engine,
    expire_on_commit=False,     # Don't expire after commit
    autoflush=False,            # Manual flush control
)

Quiz

Key Takeaways

✅ Async database sessions keep your event loop unblocked.
✅ Connection pooling reuses connections — don't create per request.
✅ Eager loading (selectinload) prevents the N+1 performance trap.
✅ Alembic migrations version your schema like you version code.
✅ Keyset pagination scales to millions of rows; OFFSET doesn't.
✅ Index strategically — index what you query, profile with EXPLAIN.

Keep coding! 🚀