Alembic Migrations

Alembic Migration Patterns ()

Database migration management with Alembic for SQLAlchemy 2.0 async applications.

Overview

• Creating or modifying database tables and columns
• Auto-generating migrations from SQLAlchemy models
• Implementing zero-downtime schema changes
• Rolling back or managing migration history
• Adding indexes on large production tables
• Setting up Alembic with async PostgreSQL (asyncpg)

Quick Reference

Initialize Alembic (Async Template)

# Initialize with async template for asyncpg
alembic init -t async migrations

# Creates:
# - alembic.ini
# - migrations/env.py (async-ready)
# - migrations/script.py.mako
# - migrations/versions/

Async env.py Configuration

# migrations/env.py
import asyncio
from logging.config import fileConfig

from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import async_engine_from_config

from alembic import context

# Import your models' Base for autogenerate
from app.models.base import Base

config = context.config
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

target_metadata = Base.metadata

def run_migrations_offline() -> None:
    """Run migrations in 'offline' mode - generates SQL."""
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )
    with context.begin_transaction():
        context.run_migrations()

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

async def run_async_migrations() -> None:
    """Run migrations in 'online' mode with async engine."""
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)

    await connectable.dispose()

def run_migrations_online() -> None:
    """Entry point for online migrations."""
    asyncio.run(run_async_migrations())

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

Migration Template

"""Add users table.

Revision ID: abc123
Revises: None
Create Date: -01-17 10:00:00.000000
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import UUID

revision = 'abc123'
down_revision = None
branch_labels = None
depends_on = None

def upgrade() -> None:
    op.create_table(
        'users',
        sa.Column('id', UUID(as_uuid=True), primary_key=True),
        sa.Column('email', sa.String(255), nullable=False),
        sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now()),
    )
    op.create_index('idx_users_email', 'users', ['email'], unique=True)

def downgrade() -> None:
    op.drop_index('idx_users_email', table_name='users')
    op.drop_table('users')

Autogenerate Migration

# Generate from model changes
alembic revision --autogenerate -m "add user preferences"

# Apply migrations
alembic upgrade head

# Rollback one step
alembic downgrade -1

# Generate SQL for review (production)
alembic upgrade head --sql > migration.sql

# Check current revision
alembic current

# Show migration history
alembic history --verbose

Running Async Code in Migrations

"""Migration with async operation.

NOTE: Alembic upgrade/downgrade cannot be async, but you can
run async code using sqlalchemy.util.await_only workaround.
"""
from alembic import op
from sqlalchemy import text
from sqlalchemy.util import await_only

def upgrade() -> None:
    # Get connection (works with async dialect)
    connection = op.get_bind()

    # For async-only operations, use await_only
    # This works because Alembic runs in greenlet context
    result = await_only(
        connection.execute(text("SELECT count(*) FROM users"))
    )

    # Standard operations work normally with async engine
    op.execute("""
        CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_users_org
        ON users (organization_id, created_at DESC)
    """)

Concurrent Index (Zero-Downtime)

def upgrade() -> None:
    # CONCURRENTLY avoids table locks on large tables
    # IMPORTANT: Cannot run inside transaction block
    op.execute("""
        CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_users_org
        ON users (organization_id, created_at DESC)
    """)

def downgrade() -> None:
    op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_users_org")

# In alembic.ini or env.py, disable transaction for this migration:
# Set transaction_per_migration = false for CONCURRENTLY operations

Two-Phase NOT NULL Migration

"""Add org_id column (phase 1 - nullable).

Phase 1: Add nullable column
Phase 2: Backfill data
Phase 3: Add NOT NULL (separate migration after verification)
"""

def upgrade() -> None:
    # Phase 1: Add as nullable first
    op.add_column('users', sa.Column('org_id', UUID(as_uuid=True), nullable=True))

    # Phase 2: Backfill with default org
    op.execute("""
        UPDATE users
        SET org_id = 'default-org-uuid'
        WHERE org_id IS NULL
    """)

    # Phase 3 in SEPARATE migration after app updated:
    # op.alter_column('users', 'org_id', nullable=False)

def downgrade() -> None:
    op.drop_column('users', 'org_id')

Key Decisions

Anti-Patterns (FORBIDDEN)

# NEVER: Add NOT NULL without default or two-phase approach
op.add_column('users', sa.Column('org_id', UUID, nullable=False))  # LOCKS TABLE, FAILS!

# NEVER: Use blocking index creation on large tables
op.create_index('idx_large', 'big_table', ['col'])  # LOCKS TABLE - use CONCURRENTLY

# NEVER: Skip downgrade implementation
def downgrade():
    pass  # WRONG - implement proper rollback

# NEVER: Modify migration after deployment
# Create a new migration instead!

# NEVER: Run migrations automatically in production
# Use: alembic upgrade head --sql > review.sql

# NEVER: Use asyncio.run() in env.py if loop exists
# Already handled by async template, but check for FastAPI lifespan conflicts

# NEVER: Run CONCURRENTLY inside transaction
op.execute("BEGIN; CREATE INDEX CONCURRENTLY ...; COMMIT;")  # FAILS

Alembic with FastAPI Lifespan

# When running migrations during FastAPI startup (advanced)
# Issue: Event loop already running

# Solution 1: Run migrations before app starts (recommended)
# In entrypoint.sh:
# alembic upgrade head && uvicorn app.main:app

# Solution 2: Use run_sync for programmatic migrations
from sqlalchemy import Connection
from alembic import command
from alembic.config import Config

async def run_migrations(connection: Connection) -> None:
    """Run migrations programmatically within existing async context."""
    def do_upgrade(connection: Connection):
        config = Config("alembic.ini")
        config.attributes["connection"] = connection
        command.upgrade(config, "head")

    await connection.run_sync(do_upgrade)

Related Skills

• database-schema-designer - Schema design and normalization patterns
• database-versioning - Version control and change management
• zero-downtime-migration - Expand/contract patterns for safe migrations
• sqlalchemy-2-async - Async SQLAlchemy session patterns
• integration-testing - Testing migrations with test databases

Capability Details

autogenerate-migrations

Keywords: autogenerate, auto-generate, revision, model sync, compare Solves:

• Auto-generate migrations from SQLAlchemy models
• Sync database with model changes
• Detect schema drift

revision-management

Keywords: upgrade, downgrade, rollback, history, current, revision Solves:

• Apply or rollback migrations
• View migration history
• Check current database version

zero-downtime-changes

Keywords: concurrent, expand contract, online migration, no downtime Solves:

• Add indexes without locking
• Rename columns safely
• Large table migrations

data-migration

Keywords: backfill, data migration, transform, batch update Solves:

• Backfill new columns with data
• Transform existing data
• Migrate between column formats

async-configuration

Keywords: asyncpg, async engine, env.py async, run_async_migrations Solves:

• Configure Alembic for async SQLAlchemy
• Run migrations with asyncpg
• Handle existing event loop conflicts

Migration Deployment Checklist

Verification steps for safe database migration deployment.

Pre-Deployment Checks

Code Review

• [ ] Migration file has descriptive docstring with purpose
• [ ] revision and down_revision are correct
• [ ] upgrade() contains all necessary changes
• [ ] downgrade() properly reverses all changes (tested!)
• [ ] No hardcoded environment-specific values
• [ ] Large table operations use CONCURRENTLY where applicable

Schema Validation

• [ ] Run alembic check - no pending model changes
• [ ] Generate SQL: alembic upgrade head --sql > migration.sql
• [ ] Review generated SQL for unexpected operations
• [ ] Verify column types match SQLAlchemy model definitions
• [ ] Check constraint names follow naming conventions

Backward Compatibility

• [ ] New columns are nullable=True or have server_default
• [ ] No column/table renames (use expand-contract pattern)
• [ ] No NOT NULL constraints added to existing columns with data
• [ ] Application code works with both old and new schema
• [ ] API responses unchanged (or versioned)

Rollback Testing

Local Rollback Verification

# Apply migration
alembic upgrade head

# Verify schema change
psql -c "\d tablename"

# Rollback migration
alembic downgrade -1

# Verify rollback complete
psql -c "\d tablename"

# Re-apply to confirm idempotency
alembic upgrade head

Rollback Checklist

• [ ] alembic downgrade -1 succeeds without errors
• [ ] Data is preserved after rollback (if applicable)
• [ ] Indexes and constraints are properly removed
• [ ] Triggers and functions are cleaned up
• [ ] Application functions correctly after rollback

Data Backup Verification

Before Production Migration

• [ ] Full database backup completed
• [ ] Backup verified (can restore to test environment)
• [ ] Point-in-time recovery configured (if using RDS/Cloud SQL)
• [ ] Backup retention policy confirmed (minimum 7 days)
• [ ] Document backup timestamp and location

Backup Commands

# PostgreSQL backup
pg_dump -Fc -v -h $DB_HOST -U $DB_USER -d $DB_NAME > backup_$(date +%Y%m%d_%H%M%S).dump

# Verify backup
pg_restore --list backup_*.dump | head -20

# Test restore to separate database
createdb restore_test
pg_restore -d restore_test backup_*.dump

Production Deployment Steps

1. Pre-Flight (T-30 minutes)

• [ ] Notify team of upcoming migration window
• [ ] Verify backup completed successfully
• [ ] Check current migration version: alembic current
• [ ] Review migration history: alembic history -v
• [ ] Confirm rollback plan documented

2. Deployment Execution

# Generate SQL for final review
alembic upgrade head --sql > /tmp/migration_$(date +%Y%m%d).sql

# Review SQL one more time
cat /tmp/migration_$(date +%Y%m%d).sql

# Apply migration with timing
time alembic upgrade head

# Verify new version
alembic current

3. Post-Migration Verification

• [ ] Check alembic current shows expected revision
• [ ] Verify schema changes with \d tablename
• [ ] Run smoke tests against API endpoints
• [ ] Check application logs for database errors
• [ ] Monitor database metrics (connections, query latency)
• [ ] Verify no increase in error rates

4. Rollback Procedure (If Needed)

# Immediate rollback
alembic downgrade -1

# Verify rollback
alembic current

# Notify team of rollback
# Investigate and fix before retry

Large Table Migration Checklist

Additional Checks for Tables > 1M Rows

• [ ] Estimated migration duration calculated
• [ ] CONCURRENTLY used for index operations
• [ ] Batch processing implemented for data migrations
• [ ] Lock wait timeout configured: SET lock_timeout = '5s'
• [ ] Statement timeout configured: SET statement_timeout = '30m'
• [ ] Maintenance window scheduled (if blocking operations)

Monitoring During Migration

• [ ] Active queries: SELECT * FROM pg_stat_activity WHERE state = 'active'
• [ ] Lock monitoring: SELECT * FROM pg_locks WHERE NOT granted
• [ ] Table bloat after migration
• [ ] Replication lag (if applicable)

Emergency Contacts

Post-Deployment Tasks

• [ ] Update documentation if schema changed significantly
• [ ] Close related tickets/issues
• [ ] Schedule VACUUM ANALYZE if large changes
• [ ] Archive migration SQL for audit trail
• [ ] Confirm monitoring alerts are not firing

Alembic Migration Examples

Complete, production-ready migration examples.

Add Column with Default Value

"""Add organization_id to users with default.

Revision ID: add_org_001
Revises: previous_rev
Create Date: 2026-01-15
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import UUID

revision = 'add_org_001'
down_revision = 'previous_rev'
branch_labels = None
depends_on = None

def upgrade() -> None:
    # Add column as nullable first (no lock)
    op.add_column('users',
        sa.Column('organization_id', UUID(as_uuid=True), nullable=True)
    )

    # Add foreign key constraint
    op.create_foreign_key(
        'fk_users_organization',
        'users', 'organizations',
        ['organization_id'], ['id'],
        ondelete='SET NULL'
    )

    # Backfill with default organization (in separate transaction for large tables)
    op.execute("""
        UPDATE users
        SET organization_id = (SELECT id FROM organizations WHERE is_default = true)
        WHERE organization_id IS NULL
    """)

def downgrade() -> None:
    op.drop_constraint('fk_users_organization', 'users', type_='foreignkey')
    op.drop_column('users', 'organization_id')

Create Index CONCURRENTLY

"""Add composite index for user search queries.

Revision ID: idx_user_search
Revises: add_org_001
Create Date: 2026-01-15

Note: Uses CONCURRENTLY to avoid blocking reads/writes.
Migration must run outside transaction.
"""
from alembic import op

revision = 'idx_user_search'
down_revision = 'add_org_001'
branch_labels = None
depends_on = None

def upgrade() -> None:
    # CRITICAL: Exit transaction for CONCURRENTLY
    op.execute("COMMIT")

    # Composite index for common query pattern
    op.execute("""
        CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_users_org_status_created
        ON users (organization_id, status, created_at DESC)
        WHERE deleted_at IS NULL
    """)

    # GIN index for full-text search on name
    op.execute("""
        CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_users_name_trgm
        ON users USING gin (name gin_trgm_ops)
    """)

def downgrade() -> None:
    op.execute("COMMIT")
    op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_users_name_trgm")
    op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_users_org_status_created")

Data Migration with Batching

"""Migrate user preferences from JSON to normalized table.

Revision ID: migrate_prefs
Revises: idx_user_search
Create Date: 2026-01-15

Handles millions of rows with batch processing.
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import UUID, JSONB
import json

revision = 'migrate_prefs'
down_revision = 'idx_user_search'
branch_labels = None
depends_on = None

BATCH_SIZE = 5000

def upgrade() -> None:
    # Create new preferences table
    op.create_table('user_preferences',
        sa.Column('id', UUID(as_uuid=True), primary_key=True,
                  server_default=sa.text('gen_random_uuid()')),
        sa.Column('user_id', UUID(as_uuid=True), nullable=False),
        sa.Column('key', sa.String(100), nullable=False),
        sa.Column('value', sa.Text, nullable=True),
        sa.Column('created_at', sa.DateTime(timezone=True),
                  server_default=sa.func.now()),
        sa.Column('updated_at', sa.DateTime(timezone=True),
                  server_default=sa.func.now()),
    )

    # Add constraints
    op.create_foreign_key('fk_user_prefs_user', 'user_preferences', 'users',
                          ['user_id'], ['id'], ondelete='CASCADE')
    op.create_unique_constraint('uq_user_prefs_user_key', 'user_preferences',
                                ['user_id', 'key'])

    # Migrate data in batches
    conn = op.get_bind()

    offset = 0
    while True:
        # Fetch batch of users with preferences
        result = conn.execute(sa.text("""
            SELECT id, preferences
            FROM users
            WHERE preferences IS NOT NULL
              AND preferences != '{}'::jsonb
            ORDER BY id
            LIMIT :batch_size OFFSET :offset
        """), {'batch_size': BATCH_SIZE, 'offset': offset})

        rows = result.fetchall()
        if not rows:
            break

        # Transform and insert preferences
        for user_id, prefs in rows:
            if prefs:
                pref_dict = prefs if isinstance(prefs, dict) else json.loads(prefs)
                for key, value in pref_dict.items():
                    conn.execute(sa.text("""
                        INSERT INTO user_preferences (user_id, key, value)
                        VALUES (:user_id, :key, :value)
                        ON CONFLICT (user_id, key) DO NOTHING
                    """), {'user_id': user_id, 'key': key, 'value': str(value)})

        conn.commit()
        offset += BATCH_SIZE
        print(f"Migrated {offset} users...")

    # Create index after data load (faster than during inserts)
    op.create_index('idx_user_prefs_user_id', 'user_preferences', ['user_id'])

def downgrade() -> None:
    # Migrate data back to JSON column
    conn = op.get_bind()

    conn.execute(sa.text("""
        UPDATE users u
        SET preferences = (
            SELECT jsonb_object_agg(key, value)
            FROM user_preferences up
            WHERE up.user_id = u.id
        )
        WHERE EXISTS (
            SELECT 1 FROM user_preferences WHERE user_id = u.id
        )
    """))
    conn.commit()

    op.drop_table('user_preferences')

Enum Type Changes

"""Add new status values to user_status enum.

Revision ID: enum_status
Revises: migrate_prefs
Create Date: 2026-01-15

PostgreSQL enum modification requires special handling.
"""
from alembic import op

revision = 'enum_status'
down_revision = 'migrate_prefs'
branch_labels = None
depends_on = None

def upgrade() -> None:
    # Add new values to existing enum
    op.execute("ALTER TYPE user_status ADD VALUE IF NOT EXISTS 'suspended'")
    op.execute("ALTER TYPE user_status ADD VALUE IF NOT EXISTS 'pending_verification'")

    # Note: Cannot remove enum values in PostgreSQL
    # For removal, must recreate enum type (see downgrade for pattern)

def downgrade() -> None:
    # Recreate enum without new values (complex operation)
    # First, update any rows using new values
    op.execute("""
        UPDATE users
        SET status = 'inactive'
        WHERE status IN ('suspended', 'pending_verification')
    """)

    # Rename old enum
    op.execute("ALTER TYPE user_status RENAME TO user_status_old")

    # Create new enum without new values
    op.execute("CREATE TYPE user_status AS ENUM ('active', 'inactive', 'deleted')")

    # Update column to use new enum
    op.execute("""
        ALTER TABLE users
        ALTER COLUMN status TYPE user_status
        USING status::text::user_status
    """)

    # Drop old enum
    op.execute("DROP TYPE user_status_old")

Add Table with Partitioning

"""Add partitioned events table for analytics.

Revision ID: events_partition
Revises: enum_status
Create Date: 2026-01-15

Uses PostgreSQL native partitioning for time-series data.
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import UUID, JSONB

revision = 'events_partition'
down_revision = 'enum_status'
branch_labels = None
depends_on = None

def upgrade() -> None:
    # Create partitioned parent table
    op.execute("""
        CREATE TABLE events (
            id UUID DEFAULT gen_random_uuid(),
            event_type VARCHAR(100) NOT NULL,
            user_id UUID,
            payload JSONB,
            created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
            PRIMARY KEY (id, created_at)
        ) PARTITION BY RANGE (created_at)
    """)

    # Create initial partitions (monthly)
    op.execute("""
        CREATE TABLE events_2026_01 PARTITION OF events
        FOR VALUES FROM ('2026-01-01') TO ('2026-02-01')
    """)
    op.execute("""
        CREATE TABLE events_2026_02 PARTITION OF events
        FOR VALUES FROM ('2026-02-01') TO ('2026-03-01')
    """)
    op.execute("""
        CREATE TABLE events_2026_03 PARTITION OF events
        FOR VALUES FROM ('2026-03-01') TO ('2026-04-01')
    """)

    # Create indexes on parent (propagates to partitions)
    op.execute("COMMIT")  # For CONCURRENTLY
    op.execute("""
        CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_events_user_type
        ON events (user_id, event_type, created_at DESC)
    """)

    # Add FK constraint
    op.create_foreign_key('fk_events_user', 'events', 'users',
                          ['user_id'], ['id'], ondelete='SET NULL')

def downgrade() -> None:
    op.drop_table('events_2026_03')
    op.drop_table('events_2026_02')
    op.drop_table('events_2026_01')
    op.drop_table('events')

Rename Column Safely (Expand-Contract)

"""Phase 1: Add email_address alongside email.

Revision ID: rename_email_p1
Revises: events_partition
Create Date: 2026-01-15

Safe column rename using expand-contract pattern.
Run Phase 2 after application code is updated.
"""
from alembic import op
import sqlalchemy as sa

revision = 'rename_email_p1'
down_revision = 'events_partition'
branch_labels = None
depends_on = None

def upgrade() -> None:
    # Add new column
    op.add_column('users', sa.Column('email_address', sa.String(255), nullable=True))

    # Copy existing data
    op.execute("UPDATE users SET email_address = email")

    # Create sync trigger for transition period
    op.execute("""
        CREATE OR REPLACE FUNCTION sync_user_email()
        RETURNS TRIGGER AS $$
        BEGIN
            IF NEW.email IS DISTINCT FROM OLD.email THEN
                NEW.email_address = NEW.email;
            ELSIF NEW.email_address IS DISTINCT FROM OLD.email_address THEN
                NEW.email = NEW.email_address;
            END IF;
            RETURN NEW;
        END;
        $$ LANGUAGE plpgsql;

        CREATE TRIGGER trg_sync_email
        BEFORE UPDATE ON users
        FOR EACH ROW EXECUTE FUNCTION sync_user_email();
    """)

def downgrade() -> None:
    op.execute("DROP TRIGGER IF EXISTS trg_sync_email ON users")
    op.execute("DROP FUNCTION IF EXISTS sync_user_email()")
    op.drop_column('users', 'email_address')

Alembic Advanced Implementation Guide

Advanced patterns for production database migrations with Alembic.

Multi-Database Migrations

Configuration Setup

# alembic/env.py - Multi-database support
from alembic import context
from sqlalchemy import engine_from_config, pool

# Define multiple databases
DATABASES = {
    'default': 'postgresql://user:pass@localhost/main',
    'analytics': 'postgresql://user:pass@localhost/analytics',
    'audit': 'postgresql://user:pass@localhost/audit',
}

def run_migrations_online():
    """Run migrations for each database."""
    for db_name, url in DATABASES.items():
        config = context.config
        config.set_main_option('sqlalchemy.url', url)

        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=get_metadata(db_name),
                version_table=f'alembic_version_{db_name}',
            )

            with context.begin_transaction():
                context.run_migrations()

Per-Database Migrations

# migrations/versions/abc123_add_analytics_table.py
"""Add analytics events table.

Revision ID: abc123
Database: analytics
"""
from alembic import op
import sqlalchemy as sa

revision = 'abc123'
down_revision = 'xyz789'
branch_labels = ('analytics',)  # Database-specific branch

def upgrade() -> None:
    op.create_table('events',
        sa.Column('id', sa.BigInteger, primary_key=True),
        sa.Column('event_type', sa.String(100), nullable=False),
        sa.Column('payload', sa.JSON, nullable=True),
        sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now()),
    )

def downgrade() -> None:
    op.drop_table('events')

Data Migrations with Batching

Batch Processing Pattern

"""Backfill user_status column with batching.

Revision ID: def456
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.orm import Session

BATCH_SIZE = 1000

def upgrade() -> None:
    # Phase 1: Add nullable column
    op.add_column('users', sa.Column('status', sa.String(20), nullable=True))

    # Phase 2: Backfill in batches
    conn = op.get_bind()
    session = Session(bind=conn)

    total_updated = 0
    while True:
        # Fetch batch of IDs without status
        result = conn.execute(sa.text("""
            SELECT id FROM users
            WHERE status IS NULL
            LIMIT :batch_size
            FOR UPDATE SKIP LOCKED
        """), {'batch_size': BATCH_SIZE})

        ids = [row[0] for row in result]
        if not ids:
            break

        # Update batch
        conn.execute(sa.text("""
            UPDATE users
            SET status = CASE
                WHEN is_active THEN 'active'
                ELSE 'inactive'
            END
            WHERE id = ANY(:ids)
        """), {'ids': ids})

        total_updated += len(ids)
        conn.commit()  # Commit per batch to release locks

        print(f"Backfilled {total_updated} rows...")

    # Phase 3: Add NOT NULL constraint (separate migration recommended)

def downgrade() -> None:
    op.drop_column('users', 'status')

Branch Management and Merging

Creating Migration Branches

# Create a feature branch
alembic revision --branch-label=feature_payments -m "start payments feature"

# Create revision on branch
alembic revision --head=feature_payments@head -m "add payment_methods table"

# View branch structure
alembic branches

# Merge branches before deployment
alembic merge feature_payments@head main@head -m "merge payments feature"

Resolving Branch Conflicts

"""Merge feature_payments and main branches.

Revision ID: merge_abc
Revises: ('abc123', 'def456')
"""
revision = 'merge_abc'
down_revision = ('abc123', 'def456')  # Tuple for merge

def upgrade() -> None:
    # No operations - just marks merge point
    pass

def downgrade() -> None:
    # Cannot downgrade past merge - requires specific branch
    raise Exception("Cannot downgrade past merge point")

Online Schema Changes (CONCURRENTLY)

Index Creation Without Locks

"""Add index concurrently on large table.

Revision ID: idx123
"""
from alembic import op

# CRITICAL: Disable transaction for CONCURRENTLY operations
def upgrade() -> None:
    # Exit transaction block
    op.execute("COMMIT")

    # Create index without locking reads/writes
    op.execute("""
        CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_orders_customer_date
        ON orders (customer_id, created_at DESC)
        WHERE status != 'cancelled'
    """)

def downgrade() -> None:
    op.execute("COMMIT")
    op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_orders_customer_date")

Column Rename (Expand-Contract Pattern)

"""Phase 1: Add new column alongside old.

Revision ID: rename_phase1
"""
from alembic import op
import sqlalchemy as sa

def upgrade() -> None:
    # Add new column
    op.add_column('users', sa.Column('full_name', sa.String(255), nullable=True))

    # Create trigger to sync during transition
    op.execute("""
        CREATE OR REPLACE FUNCTION sync_user_name()
        RETURNS TRIGGER AS $$
        BEGIN
            IF TG_OP = 'INSERT' OR TG_OP = 'UPDATE' THEN
                NEW.full_name = COALESCE(NEW.full_name, NEW.name);
                NEW.name = COALESCE(NEW.name, NEW.full_name);
            END IF;
            RETURN NEW;
        END;
        $$ LANGUAGE plpgsql;

        CREATE TRIGGER trg_sync_user_name
        BEFORE INSERT OR UPDATE ON users
        FOR EACH ROW EXECUTE FUNCTION sync_user_name();
    """)

def downgrade() -> None:
    op.execute("DROP TRIGGER IF EXISTS trg_sync_user_name ON users")
    op.execute("DROP FUNCTION IF EXISTS sync_user_name()")
    op.drop_column('users', 'full_name')

Environment-Specific Migrations

Conditional Migration Logic

"""Add analytics index (production only).

Revision ID: prod_idx
"""
from alembic import op
import os

def upgrade() -> None:
    # Only create expensive index in production
    if os.getenv('ENVIRONMENT') == 'production':
        op.execute("COMMIT")
        op.execute("""
            CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_events_timestamp
            ON events (timestamp DESC)
        """)
    else:
        # Simpler non-concurrent for dev/test
        op.create_index('idx_events_timestamp', 'events', ['timestamp'])

def downgrade() -> None:
    if os.getenv('ENVIRONMENT') == 'production':
        op.execute("COMMIT")
        op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_events_timestamp")
    else:
        op.drop_index('idx_events_timestamp', table_name='events')

Migration Hooks

Pre/Post Migration Callbacks

# alembic/env.py
from alembic import context

def before_migration(ctx, revision, heads):
    """Run before each migration."""
    print(f"Starting migration: {revision}")
    # Notify monitoring, acquire locks, etc.

def after_migration(ctx, revision, heads):
    """Run after each migration."""
    print(f"Completed migration: {revision}")
    # Clear caches, notify services, etc.

context.configure(
    # ... other config ...
    on_version_apply=after_migration,
)

Create Alembic migration: $ARGUMENTS

Migration Context (Auto-Detected)

• Current Revision: !alembic current 2>/dev/null | head -1 || echo "No current revision"
• Recent Model Changes: !git diff --name-only HEAD~5 2>/dev/null | grep -E 'models|schema' | head -5 || echo "No recent model changes detected"
• Python Version: !python --version 2>/dev/null || echo "Python 3.x"
• Alembic Version: !alembic --version 2>/dev/null || echo "Alembic not found"

Migration Template

"""$ARGUMENTS

Revision ID: !`date +%Y%m%d%H%M%S`
Revises: !`alembic current 2>/dev/null | awk '{print $1}' || echo "None"`
Create Date: !`date "+%Y-%m-%d %H:%M:%S"`

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa

# revision identifiers, used by Alembic.
revision: str = '!`date +%Y%m%d%H%M%S`'
down_revision: Union[str, None] = !`alembic current 2>/dev/null | awk '{print $1}' || echo "None"`
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    # Add your migration logic here
    pass
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    # Add your rollback logic here
    pass
    # ### end Alembic commands ###

Usage

1. Review detected model changes above 2. Fill in upgrade() with your schema changes 3. Fill in downgrade() with rollback logic 4. Run: alembic upgrade head