Chapter 10: schema evolution. SQLAlchemy doesn’t ship migrations; Alembic does. We cover the integration here; the Alembic textbook is the deep dive.

Why migrations

When your Base.metadata schema changes, the DB doesn’t auto-update. Migrations apply diffs to existing DBs.

Base.metadata.create_all(engine) works for fresh DBs in dev / tests. For production: Alembic.

Setup

uv add alembic
alembic init -t async migrations

migrations/ directory created. alembic.ini config; migrations/env.py setup.

Configuring env.py

# migrations/env.py
from src.myapp.db import Base
from src.myapp.settings import settings

config.set_main_option("sqlalchemy.url", settings.database_url)
target_metadata = Base.metadata

Tells Alembic where the metadata lives.

Autogenerate

alembic revision --autogenerate -m "add users table"

Alembic compares Base.metadata to the DB; generates a migration file with the diff.

Always review autogenerated migrations. They miss things (CHECK changes, certain index changes, default value changes).

Apply migrations

alembic upgrade head
alembic downgrade -1

head = latest. -1 = previous revision.

Migration file shape

def upgrade():
    op.create_table("users",
        sa.Column("id", sa.Integer(), primary_key=True),
        sa.Column("email", sa.String(255), nullable=False),
        sa.UniqueConstraint("email"),
    )

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

Forward + reverse. Most teams keep downgrade for dev convenience; in prod, roll forward only.

Naming conventions

For predictable constraint names:

NAMING_CONVENTION = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s",
}

class Base(DeclarativeBase):
    metadata = MetaData(naming_convention=NAMING_CONVENTION)

Without this, autogenerate may produce ambiguous names (e.g., ck_1).

Expand-contract pattern

For non-trivial schema changes (rename column, change type):

  1. Expand: add new column; backfill; code writes both; reads new with fallback.
  2. Contract: code reads/writes new only; drop old.

Each phase is its own migration + deploy. See Database Migrations 2026 .

Backfill

def upgrade():
    op.add_column("users", sa.Column("status", sa.String(20), nullable=True))
    
    # Backfill in Python (slow for big tables)
    conn = op.get_bind()
    conn.execute(text("UPDATE users SET status = 'active'"))
    
    op.alter_column("users", "status", nullable=False)

For huge tables: backfill outside Alembic in batches; alter NOT NULL when complete.

Concurrent indexes (Postgres)

def upgrade():
    op.execute("CREATE INDEX CONCURRENTLY ix_users_email ON users (email)")

Avoid table locks. Important for hot tables.

Note: CONCURRENTLY can’t run in a transaction. Use op.execute outside transaction:

# alembic.ini
[alembic]
transaction_per_migration = true

Then per-migration: op.execute(...) runs after commit.

Branching

For team workflows, branches happen:

A
├── B (your work)
└── C (their work)

Merge: alembic merge -m "merge B+C" B C

Alembic supports merge migrations. Best to keep history linear when possible.

Multiple databases

For apps with multiple databases:

alembic init -t multidb migrations

env.py iterates databases. Each gets its own version table.

Embedded migration runs

from alembic import command
from alembic.config import Config

@asynccontextmanager
async def lifespan(app):
    cfg = Config("alembic.ini")
    command.upgrade(cfg, "head")
    yield

Run on startup. Simple but blocks startup until done.

For multi-replica K8s deploys: only one should run migrations. Use a Job:

apiVersion: batch/v1
kind: Job
metadata: { name: migrate-{{ .Values.image.tag }} }
spec:
  template:
    spec:
      containers:
        - name: migrate
          image: myapp:{{ .Values.image.tag }}
          command: ["alembic", "upgrade", "head"]
      restartPolicy: OnFailure

Pre-deploy job; then rolling deploy of app.

Testing migrations

In CI:

alembic upgrade head  # apply all
pytest                # run tests against migrated DB

Or shape tests to detect drift between metadata and DB:

def test_no_drift():
    from alembic.autogenerate import compare_metadata
    from alembic.migration import MigrationContext
    
    with engine.connect() as conn:
        ctx = MigrationContext.configure(conn)
        diff = compare_metadata(ctx, Base.metadata)
        assert diff == [], f"Drift: {diff}"

Catches forgotten migrations in CI.

What Alembic doesn’t do

  • Big-table online schema changes (use pt-online-schema-change for MySQL, careful expand-contract for Postgres).
  • Data migrations at scale (do them as separate one-off jobs).
  • Cross-DB migration.

What’s next

Chapter 11: Testing.

Read this next

Building something AI-, backend-, or data-heavy and want a second pair of eyes? I do consulting and freelance work — see my projects and ways to reach me at rajpoot.dev .