Skip to main content

Migrations with Alembic

Alembic is the standard migration tool for SQLAlchemy. It tracks schema changes as versioned migration scripts — so you can apply, roll back, and audit every structural change to your database, in every environment.

Learning Focus

By the end of this lesson you can: initialize Alembic, configure it for async SQLAlchemy, generate auto-generated migration scripts, apply and roll back migrations, and integrate them into a CI/CD pipeline.

Installation and Initialization​

setup-alembic.sh
pip install alembic psycopg2-binary

# Initialize Alembic in your project
alembic init alembic

# Directory structure created:
# alembic/
#   env.py          ← Configure here
#   script.py.mako  ← Migration template
#   versions/       ← Generated migration files
# alembic.ini       ← Config file

Configuring alembic.ini​

alembic.ini
[alembic]
# Use a placeholder — we set this in env.py from environment variables
sqlalchemy.url = placeholder

script_location = alembic

Configuring alembic/env.py​

alembic/env.py
import asyncio
from logging.config import fileConfig
from sqlalchemy import pool
from sqlalchemy.ext.asyncio import create_async_engine
from alembic import context

# Import your models so Alembic sees them
from app.db.base import Base
from app.db.models import user, post  # noqa: F401
from app.core.config import settings

config = context.config
fileConfig(config.config_file_name)
target_metadata = Base.metadata

def get_url() -> str:
    return settings.DATABASE_URL

def run_migrations_offline() -> None:
    context.configure(
        url=get_url(),
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )
    with context.begin_transaction():
        context.run_migrations()

def do_run_migrations(connection) -> None:
    context.configure(connection=connection, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()

async def run_migrations_online() -> None:
    # Use synchronous URL for Alembic (asyncpg → psycopg2)
    sync_url = get_url().replace("+asyncpg", "")
    connectable = create_async_engine(sync_url, poolclass=pool.NullPool)
    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await connectable.dispose()

if context.is_offline_mode():
    run_migrations_offline()
else:
    asyncio.run(run_migrations_online())
warning

Alembic uses a sync connection for migrations even when your app uses async. The driver URL must use a sync driver (psycopg2, not asyncpg) in env.py.

Generating a Migration​

generate-migration.sh
# Auto-generate from model changes
alembic revision --autogenerate -m "create users table"

# Review the generated file in alembic/versions/
# Always review before applying — autogenerate is not perfect

Generated migration example:

alembic/versions/abc123_create_users_table.py
def upgrade() -> None:
    op.create_table(
        "users",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("username", sa.String(50), nullable=False),
        sa.Column("email", sa.String(255), nullable=False),
        sa.Column("hashed_password", sa.String(255), nullable=False),
        sa.Column("is_active", sa.Boolean(), nullable=False),
        sa.PrimaryKeyConstraint("id"),
        sa.UniqueConstraint("email"),
        sa.UniqueConstraint("username"),
    )
    op.create_index("ix_users_email", "users", ["email"])
    op.create_index("ix_users_id", "users", ["id"])

def downgrade() -> None:
    op.drop_table("users")

Applying Migrations​

apply-migrations.sh
# Apply all pending migrations
alembic upgrade head

# Apply one step at a time
alembic upgrade +1

# Check current version
alembic current

# Show migration history
alembic history --verbose

Rolling Back​

rollback-migrations.sh
# Roll back one migration
alembic downgrade -1

# Roll back to a specific revision
alembic downgrade abc123

# Roll back everything
alembic downgrade base

Production Workflow​

deploy.sh
#!/bin/bash
set -e
echo "Running database migrations..."
alembic upgrade head
echo "Starting application..."
gunicorn app.main:app -k uvicorn.workers.UvicornWorker

Common Pitfalls​

PitfallCause / SymptomFix
Autogenerate misses changesModel not imported in env.pyImport all model modules in env.py
Migration fails in productionTested only locallyAlways test migrations on a staging DB first
asyncpg URL in AlembicAlembic does not support asyncpgReplace +asyncpg with +psycopg2 in env.py
Circular import in env.pyImporting app-level objectsImport only models and settings in env.py
Duplicate constraint namesAutogenerate creates redundant constraintsReview and clean up generated migration before applying

Hands-On Practice​

migration-practice.sh
# 1) Initialize
alembic init alembic

# 2) Create your first model (User) and configure env.py

# 3) Generate the migration
alembic revision --autogenerate -m "initial schema"

# 4) Review it
cat alembic/versions/*initial*.py

# 5) Apply
alembic upgrade head

# 6) Add a column to User model
# user.py: add `bio: Mapped[str | None] = mapped_column(Text, nullable=True)`

# 7) Generate the change
alembic revision --autogenerate -m "add user bio column"

# 8) Apply
alembic upgrade head

# 9) Verify
alembic current
alembic history

What's Next​