Database Migrations with Alembic

Alembic is the database migration tool designed to work with SQLAlchemy. It tracks the history of schema changes as a series of versioned migration scripts — each script has an upgrade() function (apply the change) and a downgrade() function (revert it). When you add a column, create a table, or change an index, you generate a migration script, commit it to version control, and apply it to every environment (development, staging, production) in sequence. Alembic solves the problem of keeping your database schema in sync with your SQLAlchemy models across a team and across deployments.

Setting Up Alembic #

# Install
pip install alembic

# Initialise the migration environment (run once per project)
alembic init alembic

# Directory structure created:
# alembic/
#   env.py          -- configuration, connects to DB, imports models
#   script.py.mako  -- template for migration scripts
#   versions/       -- generated migration files
# alembic.ini       -- main configuration file

# alembic/env.py — configure to use your SQLAlchemy models
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context

# Import your models so Alembic can detect schema changes
from app.database import Base   # your DeclarativeBase
from app.models import Post, User, Tag  # all your ORM models

# Alembic Config object
config = context.config

# Set target_metadata so autogenerate can detect model changes
target_metadata = Base.metadata

# Read DATABASE_URL from environment (not hardcoded!)
import os
config.set_main_option("sqlalchemy.url", os.environ["DATABASE_URL"])

def run_migrations_offline() -> None:
    url = config.get_main_option("sqlalchemy.url")
    context.configure(url=url, target_metadata=target_metadata, literal_binds=True)
    with context.begin_transaction():
        context.run_migrations()

def run_migrations_online() -> None:
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    with connectable.connect() as connection:
        context.configure(connection=connection, target_metadata=target_metadata)
        with context.begin_transaction():
            context.run_migrations()

if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

Creating and Running Migrations #

# Generate an autogenerated migration (compares models to current DB schema)
alembic revision --autogenerate -m "add_published_at_to_posts"
# Creates: alembic/versions/abc123_add_published_at_to_posts.py

# Generate an empty migration (for manual SQL operations)
alembic revision -m "add_full_text_search_index"

# Apply all pending migrations
alembic upgrade head

# Apply one migration at a time
alembic upgrade +1

# Roll back one migration
alembic downgrade -1

# Roll back to a specific revision
alembic downgrade abc123

# Roll back all migrations
alembic downgrade base

# Show current migration version
alembic current

# Show migration history
alembic history --verbose

Migration File Structure #

# alembic/versions/abc123_add_published_at_to_posts.py
"""add published_at to posts

Revision ID: abc123def456
Revises:     xyz789abc123
Create Date: 2025-08-06 14:30:00.000000
"""
from alembic import op
import sqlalchemy as sa

revision = "abc123def456"
down_revision = "xyz789abc123"   # previous migration
branch_labels = None
depends_on = None

def upgrade() -> None:
    # ── Safe: add nullable column ─────────────────────────────────────────────
    op.add_column("posts",
        sa.Column("published_at", sa.TIMESTAMP(timezone=True), nullable=True)
    )

    # ── Create index concurrently (non-blocking on production) ───────────────
    op.execute(
        "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_posts_published_at "
        "ON posts (published_at DESC) WHERE published_at IS NOT NULL"
    )

def downgrade() -> None:
    op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_posts_published_at")
    op.drop_column("posts", "published_at")

Production Migration Workflow #

# ── Safe production migration checklist ──────────────────────────────────────

# 1. Generate migration locally
alembic revision --autogenerate -m "add_search_vector_column"

# 2. Review the generated file
cat alembic/versions/*_add_search_vector_column.py

# 3. Test on development database
alembic upgrade head
# Verify: run your test suite, check endpoints work

# 4. Test on staging
DATABASE_URL=postgresql://...staging... alembic upgrade head

# 5. Apply to production (during low-traffic window for risky changes)
DATABASE_URL=postgresql://...prod... alembic upgrade head

# 6. If something goes wrong: roll back
DATABASE_URL=postgresql://...prod... alembic downgrade -1

# ── Multiple heads (branched migrations) ─────────────────────────────────────
alembic heads     # list all head revisions (should be 1 in simple projects)
alembic merge heads   # merge two branches into one

Common Mistakes #

Mistake 1 — Hardcoding DATABASE_URL in alembic.ini #

❌ Wrong — credentials in version control:

# alembic.ini
sqlalchemy.url = postgresql://admin:s3cr3t@prod.server/mydb   # exposed!

✅ Correct — read from environment in env.py:

config.set_main_option("sqlalchemy.url", os.environ["DATABASE_URL"])   # ✓

Mistake 2 — Not reviewing autogenerated migrations #

❌ Wrong — running autogenerated migration without checking:

alembic revision --autogenerate -m "changes" && alembic upgrade head
# Autogenerate may generate: DROP TABLE users (if model was renamed!)
# Always check the script first

✅ Correct — review before applying:

alembic revision --autogenerate -m "changes"
cat alembic/versions/*_changes.py   # ✓ read it first!
alembic upgrade head

Mistake 3 — Using blocking CREATE INDEX in production migration #

❌ Wrong — locks table during index build on production:

op.create_index("idx_posts_status", "posts", ["status"])
# Blocks all reads and writes during index build!

✅ Correct — use CONCURRENTLY via op.execute():

op.execute("CREATE INDEX CONCURRENTLY idx_posts_status ON posts(status)")   # ✓

Quick Reference #