Alembic

Alembic Database Migration Management Skill

Overview

This skill provides comprehensive guidance for managing database migrations using Alembic in customer support environments. It covers everything from initial setup through complex production deployment scenarios, with a focus on maintaining data integrity and minimizing downtime for support operations.

Core Concepts

What is Alembic?

Alembic is a lightweight database migration tool for use with SQLAlchemy. It provides a way to manage changes to your database schema over time through version-controlled migration scripts. For customer support systems, this means:

• Version Control: Track all schema changes in your support database
• Reproducibility: Apply the same migrations across dev, staging, and production
• Rollback Capability: Safely revert problematic changes
• Team Collaboration: Merge schema changes from multiple developers
• Data Preservation: Migrate data during schema transformations

Migration Lifecycle in Support Systems

1. Development: Create migrations locally while developing new features 2. Testing: Validate migrations in staging environment 3. Review: Code review migration scripts before production 4. Deployment: Apply migrations to production with minimal downtime 5. Monitoring: Track migration status and handle failures 6. Rollback: Revert if issues arise in production

Installation and Initial Setup

Installing Alembic

# Install Alembic with PostgreSQL support
pip install alembic psycopg2-binary sqlalchemy

# Or add to requirements.txt
alembic>=1.13.0
sqlalchemy>=2.0.0
psycopg2-binary>=2.9.0

Initialize Alembic in Your Project

# Initialize Alembic (creates alembic/ directory and alembic.ini)
alembic init alembic

# For multiple database support
alembic init --template multidb alembic

This creates:

• alembic/: Directory containing migration scripts
• alembic/versions/: Where individual migration files live
• alembic/env.py: Migration environment configuration
• alembic.ini: Alembic configuration file

Configure Database Connection

Edit alembic.ini to set your database URL:

# For development
sqlalchemy.url = postgresql://user:password@localhost/support_dev

# For production (use environment variables)
sqlalchemy.url = postgresql://%(DB_USER)s:%(DB_PASSWORD)s@%(DB_HOST)s/%(DB_NAME)s

Better approach - use environment variables in env.py:

import os
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context

# Import your models
from myapp.models import Base

# This is the Alembic Config object
config = context.config

# Override sqlalchemy.url from environment
db_url = os.getenv('DATABASE_URL', 'postgresql://localhost/support_dev')
config.set_main_option('sqlalchemy.url', db_url)

# Set up target metadata for autogenerate
target_metadata = Base.metadata

Creating Migrations

Manual Migration Creation

Create a migration manually when you need precise control:

# Create empty migration file
alembic revision -m "add ticket priority column"

This generates a file like versions/abc123_add_ticket_priority_column.py:

"""add ticket priority column

Revision ID: abc123
Revises: def456
Create Date: 2025-01-15 10:30:00.000000
"""

from alembic import op
import sqlalchemy as sa

# revision identifiers
revision = 'abc123'
down_revision = 'def456'
branch_labels = None
depends_on = None

def upgrade() -> None:
    # Add priority column to tickets table
    op.add_column('tickets',
        sa.Column('priority', sa.String(20), nullable=True, server_default='normal')
    )

    # Create index for performance
    op.create_index('ix_tickets_priority', 'tickets', ['priority'])

def downgrade() -> None:
    # Remove index first
    op.drop_index('ix_tickets_priority', 'tickets')

    # Remove column
    op.drop_column('tickets', 'priority')

Autogenerate Migrations

Let Alembic detect schema changes automatically:

# Generate migration by comparing models to database
alembic revision --autogenerate -m "add customer satisfaction table"

Important: Always review autogenerated migrations! They may miss:

• Renamed columns (appears as drop + add)
• Changed column types requiring data conversion
• Complex constraints
• Data migrations

Example autogenerated migration:

"""add customer satisfaction table

Revision ID: xyz789
Revises: abc123
Create Date: 2025-01-15 11:00:00.000000
"""

from alembic import op
import sqlalchemy as sa

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

def upgrade() -> None:
    # Auto-generated - review before running!
    op.create_table(
        'customer_satisfaction',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('ticket_id', sa.Integer(), nullable=False),
        sa.Column('rating', sa.Integer(), nullable=False),
        sa.Column('feedback', sa.Text(), nullable=True),
        sa.Column('created_at', sa.DateTime(), nullable=False),
        sa.ForeignKeyConstraint(['ticket_id'], ['tickets.id'], ondelete='CASCADE'),
        sa.PrimaryKeyConstraint('id')
    )
    op.create_index('ix_satisfaction_ticket_id', 'customer_satisfaction', ['ticket_id'])
    op.create_index('ix_satisfaction_created_at', 'customer_satisfaction', ['created_at'])

def downgrade() -> None:
    op.drop_index('ix_satisfaction_created_at', 'customer_satisfaction')
    op.drop_index('ix_satisfaction_ticket_id', 'customer_satisfaction')
    op.drop_table('customer_satisfaction')

Data Migrations

Migrating Data During Schema Changes

When you need to transform existing data:

"""convert ticket status to new enum

Revision ID: data001
Revises: xyz789
Create Date: 2025-01-15 12:00:00.000000
"""

from alembic import op
import sqlalchemy as sa
from sqlalchemy.sql import table, column

revision = 'data001'
down_revision = 'xyz789'

def upgrade() -> None:
    # Create new status column
    op.add_column('tickets',
        sa.Column('status_new', sa.String(50), nullable=True)
    )

    # Migrate data using bulk update
    tickets = table('tickets',
        column('status', sa.String),
        column('status_new', sa.String)
    )

    # Map old statuses to new ones
    status_mapping = {
        'open': 'OPEN',
        'in_progress': 'IN_PROGRESS',
        'pending': 'WAITING_ON_CUSTOMER',
        'resolved': 'RESOLVED',
        'closed': 'CLOSED'
    }

    connection = op.get_bind()
    for old_status, new_status in status_mapping.items():
        connection.execute(
            tickets.update().where(
                tickets.c.status == old_status
            ).values(status_new=new_status)
        )

    # Make new column non-nullable now that data is migrated
    op.alter_column('tickets', 'status_new', nullable=False)

    # Drop old column and rename new one
    op.drop_column('tickets', 'status')
    op.alter_column('tickets', 'status_new', new_column_name='status')

def downgrade() -> None:
    # Reverse the migration
    op.add_column('tickets',
        sa.Column('status_old', sa.String(50), nullable=True)
    )

    tickets = table('tickets',
        column('status', sa.String),
        column('status_old', sa.String)
    )

    # Reverse mapping
    reverse_mapping = {
        'OPEN': 'open',
        'IN_PROGRESS': 'in_progress',
        'WAITING_ON_CUSTOMER': 'pending',
        'RESOLVED': 'resolved',
        'CLOSED': 'closed'
    }

    connection = op.get_bind()
    for new_status, old_status in reverse_mapping.items():
        connection.execute(
            tickets.update().where(
                tickets.c.status == new_status
            ).values(status_old=old_status)
        )

    op.alter_column('tickets', 'status_old', nullable=False)
    op.drop_column('tickets', 'status')
    op.alter_column('tickets', 'status_old', new_column_name='status')

Large Data Migrations with Batching

For large tables, process data in batches:

"""add computed resolution time to tickets

Revision ID: data002
Revises: data001
Create Date: 2025-01-15 13:00:00.000000
"""

from alembic import op
import sqlalchemy as sa
from sqlalchemy.sql import table, column, select

revision = 'data002'
down_revision = 'data001'

def upgrade() -> None:
    # Add new column
    op.add_column('tickets',
        sa.Column('resolution_time_seconds', sa.Integer(), nullable=True)
    )

    connection = op.get_bind()
    tickets = table('tickets',
        column('id', sa.Integer),
        column('created_at', sa.DateTime),
        column('resolved_at', sa.DateTime),
        column('resolution_time_seconds', sa.Integer)
    )

    # Process in batches to avoid memory issues
    batch_size = 1000
    offset = 0

    while True:
        # Get batch of tickets that need processing
        batch = connection.execute(
            select(
                tickets.c.id,
                tickets.c.created_at,
                tickets.c.resolved_at
            ).where(
                sa.and_(
                    tickets.c.resolved_at.isnot(None),
                    tickets.c.resolution_time_seconds.is_(None)
                )
            ).limit(batch_size).offset(offset)
        ).fetchall()

        if not batch:
            break

        # Update batch
        for row in batch:
            if row.resolved_at and row.created_at:
                resolution_time = (row.resolved_at - row.created_at).total_seconds()
                connection.execute(
                    tickets.update().where(
                        tickets.c.id == row.id
                    ).values(resolution_time_seconds=int(resolution_time))
                )

        offset += batch_size

    # Now make column non-nullable for future rows
    op.alter_column('tickets', 'resolution_time_seconds',
        nullable=False, server_default='0')

def downgrade() -> None:
    op.drop_column('tickets', 'resolution_time_seconds')

Running Migrations

Upgrade Database to Latest

# Upgrade to latest revision (head)
alembic upgrade head

# See what would be executed (SQL only, don't run)
alembic upgrade head --sql

# Upgrade one step at a time
alembic upgrade +1

# Upgrade to specific revision
alembic upgrade abc123

Downgrade Database

# Downgrade one revision
alembic downgrade -1

# Downgrade to specific revision
alembic downgrade abc123

# Downgrade to base (empty database)
alembic downgrade base

# Generate SQL for downgrade without executing
alembic downgrade -1 --sql

Check Current Status

# Show current database revision
alembic current

# Show current revision with details
alembic current --verbose

# Show migration history
alembic history

# Show history with current revision marked
alembic history --indicate-current

# Show specific revision range
alembic history -r base:head

Branching and Merging

Why Branch Migrations?

In customer support systems, you might have:

• Feature branches: New features developed in parallel
• Hotfix branches: Urgent fixes that can't wait for feature completion
• Team branches: Multiple teams working on different modules

Creating a Branch

# Create base for new branch
alembic revision -m "create reporting branch" \
    --head=base \
    --branch-label=reporting \
    --version-path=alembic/versions/reporting

# Add migration to specific branch
alembic revision -m "add report tables" \
    --head=reporting@head

Example branch structure:

base
├── main branch
│   ├── abc123: initial schema
│   ├── def456: add tickets
│   └── ghi789: add users
└── reporting branch
    ├── rep001: create reports table
    └── rep002: add scheduled reports

Working with Multiple Branches

# Show all branch heads
alembic heads

# Show branch points
alembic branches

# Upgrade specific branch
alembic upgrade reporting@head

# Upgrade all branches
alembic upgrade heads

Merging Branches

When features are ready to merge:

# Merge two branches
alembic merge -m "merge reporting into main" \
    main@head reporting@head

Generated merge migration:

"""merge reporting into main

Revision ID: merge001
Revises: ghi789, rep002
Create Date: 2025-01-15 14:00:00.000000
"""

from alembic import op
import sqlalchemy as sa

revision = 'merge001'
down_revision = ('ghi789', 'rep002')  # Multiple parents
branch_labels = None
depends_on = None

def upgrade() -> None:
    # Usually empty for simple merges
    # Add code if you need to reconcile conflicting changes
    pass

def downgrade() -> None:
    pass

Cross-Branch Dependencies

When one branch depends on another:

# Create migration that depends on specific revision from another branch
alembic revision -m "reporting needs user table" \
    --head=reporting@head \
    --depends-on=def456  # Revision from main branch

Testing Migrations

Unit Testing Migrations

# tests/test_migrations.py
import pytest
from alembic import command
from alembic.config import Config
from sqlalchemy import create_engine, inspect
from sqlalchemy.orm import sessionmaker

@pytest.fixture
def alembic_config():
    """Provide Alembic configuration for testing"""
    config = Config("alembic.ini")
    config.set_main_option(
        "sqlalchemy.url",
        "postgresql://localhost/support_test"
    )
    return config

@pytest.fixture
def test_db(alembic_config):
    """Create test database and apply migrations"""
    # Create engine
    engine = create_engine(
        alembic_config.get_main_option("sqlalchemy.url")
    )

    # Run migrations to head
    command.upgrade(alembic_config, "head")

    yield engine

    # Cleanup - downgrade to base
    command.downgrade(alembic_config, "base")
    engine.dispose()

def test_migration_creates_tickets_table(test_db):
    """Test that migrations create expected tables"""
    inspector = inspect(test_db)
    tables = inspector.get_table_names()

    assert 'tickets' in tables
    assert 'users' in tables
    assert 'customer_satisfaction' in tables

def test_tickets_table_structure(test_db):
    """Test ticket table has correct columns"""
    inspector = inspect(test_db)
    columns = {col['name']: col for col in inspector.get_columns('tickets')}

    assert 'id' in columns
    assert 'priority' in columns
    assert 'status' in columns
    assert 'created_at' in columns
    assert 'resolution_time_seconds' in columns

    # Check column types
    assert columns['priority']['type'].python_type == str
    assert columns['status']['type'].python_type == str

def test_migration_upgrade_downgrade_cycle(alembic_config):
    """Test that upgrade -> downgrade -> upgrade works"""
    # Start at base
    command.downgrade(alembic_config, "base")

    # Upgrade to head
    command.upgrade(alembic_config, "head")

    # Downgrade one step
    command.downgrade(alembic_config, "-1")

    # Upgrade back to head
    command.upgrade(alembic_config, "head")

    # Should complete without errors

def test_data_migration_preserves_data(test_db):
    """Test that data migrations don't lose data"""
    from sqlalchemy.orm import sessionmaker
    from myapp.models import Ticket

    Session = sessionmaker(bind=test_db)
    session = Session()

    # Insert test data
    ticket = Ticket(
        title="Test ticket",
        status="OPEN",
        priority="high"
    )
    session.add(ticket)
    session.commit()
    ticket_id = ticket.id
    session.close()

    # Run a migration that modifies tickets table
    # (This would be a specific revision)
    # command.upgrade(alembic_config, "specific_revision")

    # Verify data still exists
    session = Session()
    retrieved = session.query(Ticket).filter_by(id=ticket_id).first()
    assert retrieved is not None
    assert retrieved.title == "Test ticket"
    session.close()

Integration Testing

# tests/test_migration_integration.py
import pytest
from alembic import command
from alembic.config import Config
from alembic.script import ScriptDirectory
from alembic.runtime.migration import MigrationContext

def test_no_pending_migrations(alembic_config, test_db):
    """Ensure all migrations are applied in test environment"""
    script = ScriptDirectory.from_config(alembic_config)

    with test_db.connect() as connection:
        context = MigrationContext.configure(connection)
        current_heads = set(context.get_current_heads())
        script_heads = set(script.get_heads())

        assert current_heads == script_heads, \
            f"Database has pending migrations. Current: {current_heads}, Expected: {script_heads}"

def test_migration_order_is_valid(alembic_config):
    """Verify migration chain has no gaps or conflicts"""
    script = ScriptDirectory.from_config(alembic_config)

    # Get all revisions
    revisions = list(script.walk_revisions())

    # Check each revision has valid down_revision
    for revision in revisions:
        if revision.down_revision is not None:
            if isinstance(revision.down_revision, tuple):
                # Merge point
                for down_rev in revision.down_revision:
                    assert script.get_revision(down_rev) is not None
            else:
                assert script.get_revision(revision.down_revision) is not None

def test_check_command_detects_drift(alembic_config, test_db):
    """Test that check command detects schema drift"""
    # This test verifies that `alembic check` works correctly
    try:
        command.check(alembic_config)
        # If no exception, database matches models
        assert True
    except Exception as e:
        # If exception, there's drift between DB and models
        pytest.fail(f"Schema drift detected: {e}")

Testing Migration Performance

# tests/test_migration_performance.py
import time
import pytest
from alembic import command

def test_migration_completes_within_time_limit(alembic_config):
    """Ensure migrations complete within acceptable time"""
    # Downgrade to base
    command.downgrade(alembic_config, "base")

    # Time the upgrade
    start = time.time()
    command.upgrade(alembic_config, "head")
    duration = time.time() - start

    # Assert completes within 60 seconds
    assert duration < 60, f"Migration took {duration}s, exceeds 60s limit"

@pytest.mark.slow
def test_data_migration_with_large_dataset(alembic_config, test_db):
    """Test data migration performance with realistic data volume"""
    from sqlalchemy.orm import sessionmaker
    from myapp.models import Ticket

    Session = sessionmaker(bind=test_db)
    session = Session()

    # Create 10,000 test tickets
    tickets = [
        Ticket(
            title=f"Test ticket {i}",
            status="OPEN",
            priority="normal"
        )
        for i in range(10000)
    ]
    session.bulk_save_objects(tickets)
    session.commit()
    session.close()

    # Run data migration and measure time
    start = time.time()
    command.upgrade(alembic_config, "data002")  # Specific data migration
    duration = time.time() - start

    # Should process 10k records in reasonable time
    assert duration < 30, f"Data migration took {duration}s for 10k records"

CI/CD Integration

GitHub Actions Workflow

# .github/workflows/migrations.yml
name: Database Migrations

on:
  pull_request:
    paths:
      - 'alembic/versions/**'
      - 'myapp/models/**'
      - 'alembic.ini'
      - 'alembic/env.py'
  push:
    branches:
      - main
      - develop

jobs:
  test-migrations:
    runs-on: ubuntu-latest

    services:
      postgres:
        image: postgres:15
        env:
          POSTGRES_PASSWORD: postgres
          POSTGRES_DB: support_test
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
        ports:
          - 5432:5432

    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install pytest pytest-cov

      - name: Run migration tests
        env:
          DATABASE_URL: postgresql://postgres:postgres@localhost/support_test
        run: |
          # Test upgrade to head
          alembic upgrade head

          # Test downgrade to base
          alembic downgrade base

          # Test upgrade again
          alembic upgrade head

          # Run pytest for migration tests
          pytest tests/test_migrations.py -v

      - name: Check for schema drift
        env:
          DATABASE_URL: postgresql://postgres:postgres@localhost/support_test
        run: |
          alembic check

      - name: Validate migration history
        run: |
          # Check for multiple heads (should be only one)
          HEADS_COUNT=$(alembic heads | wc -l)
          if [ "$HEADS_COUNT" -gt 1 ]; then
            echo "ERROR: Multiple heads detected. Please merge branches."
            alembic heads
            exit 1
          fi

  review-migration-sql:
    runs-on: ubuntu-latest
    if: github.event_name == 'pull_request'

    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: pip install -r requirements.txt

      - name: Generate SQL for review
        run: |
          # Generate SQL without executing
          alembic upgrade head --sql > migration.sql

      - name: Upload SQL artifact
        uses: actions/upload-artifact@v3
        with:
          name: migration-sql
          path: migration.sql

      - name: Comment PR with SQL
        uses: actions/github-script@v6
        with:
          script: |
            const fs = require('fs');
            const sql = fs.readFileSync('migration.sql', 'utf8');

            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: `## Migration SQL\n\n\`\`\`sql\n${sql}\n\`\`\``
            });

Deployment Script

#!/bin/bash
# scripts/deploy_migrations.sh

set -e  # Exit on error

echo "Starting database migration deployment..."

# Environment variables
DB_HOST="${DB_HOST:-localhost}"
DB_NAME="${DB_NAME:-support_prod}"
DB_USER="${DB_USER:-postgres}"
DATABASE_URL="postgresql://${DB_USER}:${DB_PASSWORD}@${DB_HOST}/${DB_NAME}"

# Configuration
BACKUP_DIR="./backups"
TIMESTAMP=$(date +%Y%m%d_%H%M%S)
BACKUP_FILE="${BACKUP_DIR}/pre_migration_${TIMESTAMP}.sql"

# Create backup directory
mkdir -p "$BACKUP_DIR"

# 1. Backup database before migration
echo "Creating database backup..."
pg_dump "$DATABASE_URL" > "$BACKUP_FILE"
echo "Backup created: $BACKUP_FILE"

# 2. Check current migration status
echo "Current migration status:"
alembic current

# 3. Show pending migrations
echo "Pending migrations:"
alembic history --verbose | grep -A 5 "head"

# 4. Run migrations with timeout
echo "Running migrations..."
timeout 300 alembic upgrade head || {
    echo "ERROR: Migration failed or timed out!"
    echo "Restoring from backup..."
    psql "$DATABASE_URL" < "$BACKUP_FILE"
    exit 1
}

# 5. Verify migration success
echo "Verifying migration status..."
CURRENT_REV=$(alembic current | grep "Rev:" | awk '{print $2}')
HEAD_REV=$(alembic heads | awk '{print $1}')

if [ "$CURRENT_REV" != "$HEAD_REV" ]; then
    echo "ERROR: Migration incomplete. Current: $CURRENT_REV, Expected: $HEAD_REV"
    echo "Restoring from backup..."
    psql "$DATABASE_URL" < "$BACKUP_FILE"
    exit 1
fi

echo "Migration completed successfully!"
echo "Current revision: $CURRENT_REV"

# 6. Cleanup old backups (keep last 10)
echo "Cleaning up old backups..."
ls -t "$BACKUP_DIR"/*.sql | tail -n +11 | xargs -r rm

echo "Deployment complete!"

Production Best Practices

Pre-Deployment Checklist

• [ ] Migration tested in development environment
• [ ] Migration tested in staging with production-like data
• [ ] Migration reviewed by at least one team member
• [ ] Downgrade path tested and verified
• [ ] Performance impact assessed for large tables
• [ ] Database backup plan in place
• [ ] Rollback procedure documented
• [ ] Maintenance window scheduled (if needed)
• [ ] Team notified of deployment
• [ ] Monitoring alerts configured

Zero-Downtime Migrations

For critical support systems that can't go offline:

Phase 1: Additive Changes

"""add new column (phase 1)

Revision ID: zd001
"""

def upgrade() -> None:
    # Add new column as nullable
    op.add_column('tickets',
        sa.Column('new_field', sa.String(100), nullable=True)
    )

def downgrade() -> None:
    op.drop_column('tickets', 'new_field')

Phase 2: Data Migration (Background)

"""populate new column (phase 2)

Revision ID: zd002
"""

def upgrade() -> None:
    # Update in small batches during low-traffic periods
    connection = op.get_bind()

    batch_size = 100
    while True:
        result = connection.execute(
            """
            UPDATE tickets
            SET new_field = calculate_value(old_field)
            WHERE new_field IS NULL
            LIMIT {batch_size}
            """.format(batch_size=batch_size)
        )
        if result.rowcount == 0:
            break

        # Small delay to reduce database load
        import time
        time.sleep(0.1)

def downgrade() -> None:
    connection = op.get_bind()
    connection.execute("UPDATE tickets SET new_field = NULL")

Phase 3: Make Required

"""make new column required (phase 3)

Revision ID: zd003
"""

def upgrade() -> None:
    # Now that all rows have values, make it non-nullable
    op.alter_column('tickets', 'new_field',
        nullable=False,
        server_default='default_value'
    )

def downgrade() -> None:
    op.alter_column('tickets', 'new_field',
        nullable=True,
        server_default=None
    )

Phase 4: Remove Old Column (Optional)

"""remove old column (phase 4)

Revision ID: zd004
"""

def upgrade() -> None:
    op.drop_column('tickets', 'old_field')

def downgrade() -> None:
    op.add_column('tickets',
        sa.Column('old_field', sa.String(100), nullable=True)
    )

Handling Migration Failures

# alembic/env.py additions for error handling

from alembic import context
import logging

logger = logging.getLogger('alembic.env')

def run_migrations_online():
    """Run migrations in 'online' mode with error handling"""

    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,
            transaction_per_migration=True,  # Rollback individual migrations
            compare_type=True,
            compare_server_default=True
        )

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

        except Exception as e:
            logger.error(f"Migration failed: {e}")
            logger.error("Rolling back transaction...")
            # Transaction automatically rolled back
            raise

        else:
            logger.info("Migration completed successfully")

Advanced Configuration

Custom Migration Template

Create custom template for your organization:

# alembic/script.py.mako
"""${message}

Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}

Author: ${author if author else 'Support Team'}
Jira: ${jira_ticket if jira_ticket else 'N/A'}
"""

from alembic import op
import sqlalchemy as sa
${imports if imports else ""}

# revision identifiers, used by Alembic.
revision = ${repr(up_revision)}
down_revision = ${repr(down_revision)}
branch_labels = ${repr(branch_labels)}
depends_on = ${repr(depends_on)}


def upgrade() -> None:
    """Apply migration changes"""
    ${upgrades if upgrades else "pass"}


def downgrade() -> None:
    """Revert migration changes"""
    ${downgrades if downgrades else "pass"}

Multi-Database Support

For systems with separate databases (e.g., main DB + analytics):

# alembic/env.py for multiple databases

def run_migrations_online():
    """Run migrations for multiple databases"""

    # Configuration for each database
    engines = {
        'main': {
            'url': os.getenv('MAIN_DB_URL'),
            'target_metadata': main_metadata
        },
        'analytics': {
            'url': os.getenv('ANALYTICS_DB_URL'),
            'target_metadata': analytics_metadata
        }
    }

    for name, config in engines.items():
        logger.info(f"Running migrations for {name} database")

        engine = create_engine(config['url'])

        with engine.connect() as connection:
            context.configure(
                connection=connection,
                target_metadata=config['target_metadata'],
                upgrade_token=f"{name}_upgrade",
                downgrade_token=f"{name}_downgrade"
            )

            with context.begin_transaction():
                context.run_migrations(engine_name=name)

Troubleshooting

Common Issues and Solutions

Multiple Heads Error

# Problem: "Multiple heads exist"
# Solution: Merge the branches
alembic merge heads -m "merge branches"

Migration Out of Sync

# Problem: Database revision doesn't match migration history
# Solution: Stamp database to specific revision
alembic stamp head

# Or stamp to specific revision
alembic stamp abc123

Failed Migration Cleanup

# Problem: Migration failed midway
# Solution: Manual cleanup

# 1. Check current state
alembic current

# 2. Manually fix database issues
psql $DATABASE_URL

# 3. Stamp to correct revision
alembic stamp previous_working_revision

# 4. Try migration again
alembic upgrade head

Circular Dependencies

# Problem: "Circular dependency detected"
# Solution: Use depends_on instead of down_revision
alembic revision -m "fix circular dependency" \
    --head=branch_a@head \
    --depends-on=branch_b_revision

Summary

This skill covered comprehensive Alembic usage for customer support systems:

1. Setup: Installation, configuration, and initialization 2. Creating Migrations: Manual and autogenerated approaches 3. Data Migrations: Transforming data during schema changes 4. Running Migrations: Upgrade, downgrade, and status commands 5. Branching: Managing parallel development streams 6. Testing: Unit, integration, and performance testing 7. CI/CD: Automation and deployment strategies 8. Production: Zero-downtime migrations and best practices 9. Advanced: Custom templates and multi-database support 10. Troubleshooting: Common issues and solutions

Always remember:

• Review autogenerated migrations
• Test migrations thoroughly before production
• Keep backups before major migrations
• Plan for rollback scenarios
• Monitor migration performance
• Document complex migrations

For more examples, see EXAMPLES.md in this skill package.

Alembic Migration Examples for Customer Support Systems

This document provides 15+ practical, runnable examples for common database migration scenarios in customer support environments. Each example includes complete code that you can adapt for your own use cases.

Table of Contents

1. Initial Database Setup 2. Adding Columns to Existing Table 3. Creating Performance Indexes 4. Adding Tables with Foreign Keys 5. Modifying Column Types Safely 6. Data Migration for Status Values 7. Autogenerate from SQLAlchemy Models 8. Complex Manual Migration 9. Downgrade Procedures 10. Creating Migration Branches 11. Merging Migration Branches 12. Online Migration with Minimal Downtime 13. Testing Migrations with Pytest 14. Rolling Back Failed Migrations 15. Production Deployment Workflow 16. Batch Data Migration 17. Adding Enums and Constraints 18. Multi-Table Data Migration

---

Example 1: Initial Database Setup

Scenario: Setting up the initial schema for a customer support ticketing system.

Command:

alembic revision -m "create initial support schema"

Migration File (versions/001_create_initial_support_schema.py):

"""create initial support schema

Revision ID: 001_initial
Revises:
Create Date: 2025-01-15 10:00:00.000000
"""

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql

revision = '001_initial'
down_revision = None
branch_labels = None
depends_on = None


def upgrade() -> None:
    # Create users table
    op.create_table(
        'users',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('email', sa.String(255), nullable=False, unique=True),
        sa.Column('full_name', sa.String(200), nullable=False),
        sa.Column('role', sa.String(50), nullable=False, server_default='agent'),
        sa.Column('is_active', sa.Boolean(), nullable=False, server_default='true'),
        sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.text('NOW()')),
        sa.Column('updated_at', sa.DateTime(), nullable=False, server_default=sa.text('NOW()'))
    )

    # Create customers table
    op.create_table(
        'customers',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('email', sa.String(255), nullable=False, unique=True),
        sa.Column('name', sa.String(200), nullable=False),
        sa.Column('company', sa.String(200), nullable=True),
        sa.Column('phone', sa.String(50), nullable=True),
        sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.text('NOW()')),
        sa.Column('updated_at', sa.DateTime(), nullable=False, server_default=sa.text('NOW()'))
    )

    # Create tickets table
    op.create_table(
        'tickets',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('customer_id', sa.Integer(), nullable=False),
        sa.Column('assigned_user_id', sa.Integer(), nullable=True),
        sa.Column('subject', sa.String(500), nullable=False),
        sa.Column('description', sa.Text(), nullable=False),
        sa.Column('status', sa.String(50), nullable=False, server_default='open'),
        sa.Column('priority', sa.String(20), nullable=False, server_default='normal'),
        sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.text('NOW()')),
        sa.Column('updated_at', sa.DateTime(), nullable=False, server_default=sa.text('NOW()')),
        sa.Column('resolved_at', sa.DateTime(), nullable=True),
        sa.ForeignKeyConstraint(['customer_id'], ['customers.id'], ondelete='CASCADE'),
        sa.ForeignKeyConstraint(['assigned_user_id'], ['users.id'], ondelete='SET NULL')
    )

    # Create ticket comments table
    op.create_table(
        'ticket_comments',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('ticket_id', sa.Integer(), nullable=False),
        sa.Column('user_id', sa.Integer(), nullable=True),
        sa.Column('customer_id', sa.Integer(), nullable=True),
        sa.Column('content', sa.Text(), nullable=False),
        sa.Column('is_internal', sa.Boolean(), nullable=False, server_default='false'),
        sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.text('NOW()')),
        sa.ForeignKeyConstraint(['ticket_id'], ['tickets.id'], ondelete='CASCADE'),
        sa.ForeignKeyConstraint(['user_id'], ['users.id'], ondelete='SET NULL'),
        sa.ForeignKeyConstraint(['customer_id'], ['customers.id'], ondelete='SET NULL')
    )

    # Create basic indexes
    op.create_index('ix_users_email', 'users', ['email'])
    op.create_index('ix_users_role', 'users', ['role'])
    op.create_index('ix_customers_email', 'customers', ['email'])
    op.create_index('ix_tickets_status', 'tickets', ['status'])
    op.create_index('ix_tickets_priority', 'tickets', ['priority'])
    op.create_index('ix_tickets_customer_id', 'tickets', ['customer_id'])
    op.create_index('ix_tickets_assigned_user_id', 'tickets', ['assigned_user_id'])
    op.create_index('ix_tickets_created_at', 'tickets', ['created_at'])
    op.create_index('ix_comments_ticket_id', 'ticket_comments', ['ticket_id'])


def downgrade() -> None:
    # Drop indexes
    op.drop_index('ix_comments_ticket_id', 'ticket_comments')
    op.drop_index('ix_tickets_created_at', 'tickets')
    op.drop_index('ix_tickets_assigned_user_id', 'tickets')
    op.drop_index('ix_tickets_customer_id', 'tickets')
    op.drop_index('ix_tickets_priority', 'tickets')
    op.drop_index('ix_tickets_status', 'tickets')
    op.drop_index('ix_customers_email', 'customers')
    op.drop_index('ix_users_role', 'users')
    op.drop_index('ix_users_email', 'users')

    # Drop tables in reverse order
    op.drop_table('ticket_comments')
    op.drop_table('tickets')
    op.drop_table('customers')
    op.drop_table('users')

Run the migration:

alembic upgrade head

---

Example 2: Adding Columns to Existing Table

Scenario: Adding SLA tracking fields to the tickets table.

Command:

alembic revision -m "add sla fields to tickets"

Migration File (versions/002_add_sla_fields_to_tickets.py):

"""add sla fields to tickets

Revision ID: 002_sla_fields
Revises: 001_initial
Create Date: 2025-01-15 11:00:00.000000
"""

from alembic import op
import sqlalchemy as sa

revision = '002_sla_fields'
down_revision = '001_initial'
branch_labels = None
depends_on = None


def upgrade() -> None:
    # Add SLA deadline column
    op.add_column(
        'tickets',
        sa.Column('sla_deadline', sa.DateTime(), nullable=True)
    )

    # Add SLA violated flag
    op.add_column(
        'tickets',
        sa.Column('sla_violated', sa.Boolean(), nullable=False, server_default='false')
    )

    # Add first response time in seconds
    op.add_column(
        'tickets',
        sa.Column('first_response_time_seconds', sa.Integer(), nullable=True)
    )

    # Add resolution time in seconds
    op.add_column(
        'tickets',
        sa.Column('resolution_time_seconds', sa.Integer(), nullable=True)
    )

    # Create index for SLA queries
    op.create_index('ix_tickets_sla_deadline', 'tickets', ['sla_deadline'])
    op.create_index('ix_tickets_sla_violated', 'tickets', ['sla_violated'])


def downgrade() -> None:
    # Drop indexes
    op.drop_index('ix_tickets_sla_violated', 'tickets')
    op.drop_index('ix_tickets_sla_deadline', 'tickets')

    # Drop columns
    op.drop_column('tickets', 'resolution_time_seconds')
    op.drop_column('tickets', 'first_response_time_seconds')
    op.drop_column('tickets', 'sla_violated')
    op.drop_column('tickets', 'sla_deadline')

Run the migration:

alembic upgrade head

---

Example 3: Creating Performance Indexes

Scenario: Adding composite indexes for common queries in the support dashboard.

Command:

alembic revision -m "add performance indexes for dashboard"

Migration File (versions/003_add_performance_indexes.py):

"""add performance indexes for dashboard

Revision ID: 003_perf_indexes
Revises: 002_sla_fields
Create Date: 2025-01-15 12:00:00.000000
"""

from alembic import op
import sqlalchemy as sa

revision = '003_perf_indexes'
down_revision = '002_sla_fields'
branch_labels = None
depends_on = None


def upgrade() -> None:
    # Composite index for agent dashboard (status + assigned user)
    op.create_index(
        'ix_tickets_status_assigned_user',
        'tickets',
        ['status', 'assigned_user_id']
    )

    # Composite index for priority + status queries
    op.create_index(
        'ix_tickets_priority_status',
        'tickets',
        ['priority', 'status']
    )

    # Composite index for customer ticket history
    op.create_index(
        'ix_tickets_customer_created',
        'tickets',
        ['customer_id', 'created_at'],
        postgresql_using='btree'
    )

    # Partial index for open tickets only (faster queries)
    op.create_index(
        'ix_tickets_open_created',
        'tickets',
        ['created_at'],
        postgresql_where=sa.text("status IN ('open', 'in_progress')")
    )

    # Partial index for unassigned tickets
    op.create_index(
        'ix_tickets_unassigned',
        'tickets',
        ['created_at'],
        postgresql_where=sa.text("assigned_user_id IS NULL")
    )

    # Text search index for ticket subjects (PostgreSQL)
    op.execute("""
        CREATE INDEX ix_tickets_subject_fulltext
        ON tickets
        USING gin(to_tsvector('english', subject))
    """)


def downgrade() -> None:
    # Drop indexes
    op.execute("DROP INDEX IF EXISTS ix_tickets_subject_fulltext")
    op.drop_index('ix_tickets_unassigned', 'tickets')
    op.drop_index('ix_tickets_open_created', 'tickets')
    op.drop_index('ix_tickets_customer_created', 'tickets')
    op.drop_index('ix_tickets_priority_status', 'tickets')
    op.drop_index('ix_tickets_status_assigned_user', 'tickets')

Run the migration:

alembic upgrade head

---

Example 4: Adding Tables with Foreign Keys

Scenario: Adding a table to track customer satisfaction surveys.

Command:

alembic revision -m "create customer satisfaction table"

Migration File (versions/004_create_satisfaction_table.py):

"""create customer satisfaction table

Revision ID: 004_satisfaction
Revises: 003_perf_indexes
Create Date: 2025-01-15 13:00:00.000000
"""

from alembic import op
import sqlalchemy as sa

revision = '004_satisfaction'
down_revision = '003_perf_indexes'
branch_labels = None
depends_on = None


def upgrade() -> None:
    # Create satisfaction surveys table
    op.create_table(
        'satisfaction_surveys',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('ticket_id', sa.Integer(), nullable=False),
        sa.Column('customer_id', sa.Integer(), nullable=False),
        sa.Column('rating', sa.Integer(), nullable=False),
        sa.Column('feedback', sa.Text(), nullable=True),
        sa.Column('survey_sent_at', sa.DateTime(), nullable=False),
        sa.Column('survey_completed_at', sa.DateTime(), nullable=False, server_default=sa.text('NOW()')),
        sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.text('NOW()')),

        # Foreign key constraints
        sa.ForeignKeyConstraint(
            ['ticket_id'],
            ['tickets.id'],
            ondelete='CASCADE',
            name='fk_surveys_ticket'
        ),
        sa.ForeignKeyConstraint(
            ['customer_id'],
            ['customers.id'],
            ondelete='CASCADE',
            name='fk_surveys_customer'
        ),

        # Check constraint for valid ratings
        sa.CheckConstraint(
            'rating >= 1 AND rating <= 5',
            name='ck_surveys_rating_range'
        )
    )

    # Create indexes
    op.create_index('ix_surveys_ticket_id', 'satisfaction_surveys', ['ticket_id'])
    op.create_index('ix_surveys_customer_id', 'satisfaction_surveys', ['customer_id'])
    op.create_index('ix_surveys_rating', 'satisfaction_surveys', ['rating'])
    op.create_index('ix_surveys_completed_at', 'satisfaction_surveys', ['survey_completed_at'])

    # Add unique constraint (one survey per ticket)
    op.create_unique_constraint(
        'uq_surveys_ticket',
        'satisfaction_surveys',
        ['ticket_id']
    )


def downgrade() -> None:
    # Drop table (foreign keys and constraints drop automatically)
    op.drop_table('satisfaction_surveys')

Run the migration:

alembic upgrade head

---

Example 5: Modifying Column Types Safely

Scenario: Converting ticket priority from string to enum and increasing subject length.

Command:

alembic revision -m "modify ticket column types"

Migration File (versions/005_modify_ticket_column_types.py):

"""modify ticket column types

Revision ID: 005_modify_types
Revises: 004_satisfaction
Create Date: 2025-01-15 14:00:00.000000
"""

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql

revision = '005_modify_types'
down_revision = '004_satisfaction'
branch_labels = None
depends_on = None


def upgrade() -> None:
    # Step 1: Create enum type for priority
    priority_enum = postgresql.ENUM(
        'low', 'normal', 'high', 'urgent',
        name='ticket_priority_enum',
        create_type=True
    )
    priority_enum.create(op.get_bind(), checkfirst=True)

    # Step 2: Add new column with enum type
    op.add_column(
        'tickets',
        sa.Column('priority_new', priority_enum, nullable=True)
    )

    # Step 3: Migrate data from old column to new
    op.execute("""
        UPDATE tickets
        SET priority_new = CAST(priority AS ticket_priority_enum)
        WHERE priority IN ('low', 'normal', 'high', 'urgent')
    """)

    # Step 4: Handle any invalid values (set to 'normal')
    op.execute("""
        UPDATE tickets
        SET priority_new = 'normal'::ticket_priority_enum
        WHERE priority_new IS NULL
    """)

    # Step 5: Make new column non-nullable
    op.alter_column('tickets', 'priority_new', nullable=False)

    # Step 6: Drop old column and rename new one
    op.drop_index('ix_tickets_priority', 'tickets')
    op.drop_column('tickets', 'priority')
    op.alter_column('tickets', 'priority_new', new_column_name='priority')

    # Step 7: Recreate index
    op.create_index('ix_tickets_priority', 'tickets', ['priority'])

    # Step 8: Increase subject length from 500 to 1000
    op.alter_column(
        'tickets',
        'subject',
        type_=sa.String(1000),
        existing_type=sa.String(500),
        existing_nullable=False
    )


def downgrade() -> None:
    # Reverse subject length change
    op.alter_column(
        'tickets',
        'subject',
        type_=sa.String(500),
        existing_type=sa.String(1000),
        existing_nullable=False
    )

    # Convert enum back to string
    op.drop_index('ix_tickets_priority', 'tickets')

    op.add_column(
        'tickets',
        sa.Column('priority_old', sa.String(20), nullable=True)
    )

    op.execute("""
        UPDATE tickets
        SET priority_old = CAST(priority AS VARCHAR)
    """)

    op.alter_column('tickets', 'priority_old', nullable=False)
    op.drop_column('tickets', 'priority')
    op.alter_column('tickets', 'priority_old', new_column_name='priority')

    op.create_index('ix_tickets_priority', 'tickets', ['priority'])

    # Drop enum type
    priority_enum = postgresql.ENUM(
        'low', 'normal', 'high', 'urgent',
        name='ticket_priority_enum'
    )
    priority_enum.drop(op.get_bind(), checkfirst=True)

Run the migration:

alembic upgrade head

---

Example 6: Data Migration for Status Values

Scenario: Migrating ticket statuses to a new standardized format.

Command:

alembic revision -m "standardize ticket status values"

Migration File (versions/006_standardize_status_values.py):

"""standardize ticket status values

Revision ID: 006_status_migration
Revises: 005_modify_types
Create Date: 2025-01-15 15:00:00.000000
"""

from alembic import op
import sqlalchemy as sa
from sqlalchemy.sql import table, column

revision = '006_status_migration'
down_revision = '005_modify_types'
branch_labels = None
depends_on = None


def upgrade() -> None:
    # Define table structure for data migration
    tickets = table(
        'tickets',
        column('id', sa.Integer),
        column('status', sa.String)
    )

    # Mapping of old status values to new standardized values
    status_mapping = {
        'open': 'OPEN',
        'new': 'OPEN',
        'in_progress': 'IN_PROGRESS',
        'working': 'IN_PROGRESS',
        'pending': 'WAITING_ON_CUSTOMER',
        'waiting': 'WAITING_ON_CUSTOMER',
        'customer_reply': 'WAITING_ON_CUSTOMER',
        'resolved': 'RESOLVED',
        'fixed': 'RESOLVED',
        'closed': 'CLOSED',
        'done': 'CLOSED'
    }

    connection = op.get_bind()

    # Update each old status to new status
    for old_status, new_status in status_mapping.items():
        connection.execute(
            tickets.update()
            .where(tickets.c.status == old_status)
            .values(status=new_status)
        )

    # Handle any remaining unmapped statuses (set to OPEN)
    valid_statuses = set(status_mapping.values())
    connection.execute(
        tickets.update()
        .where(~tickets.c.status.in_(valid_statuses))
        .values(status='OPEN')
    )

    # Add check constraint to ensure only valid statuses
    op.create_check_constraint(
        'ck_tickets_status_valid',
        'tickets',
        sa.text("status IN ('OPEN', 'IN_PROGRESS', 'WAITING_ON_CUSTOMER', 'RESOLVED', 'CLOSED')")
    )


def downgrade() -> None:
    # Remove check constraint
    op.drop_constraint('ck_tickets_status_valid', 'tickets', type_='check')

    # Reverse mapping (new to old - using most common old value)
    tickets = table(
        'tickets',
        column('status', sa.String)
    )

    reverse_mapping = {
        'OPEN': 'open',
        'IN_PROGRESS': 'in_progress',
        'WAITING_ON_CUSTOMER': 'pending',
        'RESOLVED': 'resolved',
        'CLOSED': 'closed'
    }

    connection = op.get_bind()

    for new_status, old_status in reverse_mapping.items():
        connection.execute(
            tickets.update()
            .where(tickets.c.status == new_status)
            .values(status=old_status)
        )

Run the migration:

alembic upgrade head

---

Example 7: Autogenerate from SQLAlchemy Models

Scenario: Using autogenerate to create migration from model changes.

SQLAlchemy Model (myapp/models.py):

from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from datetime import datetime

Base = declarative_base()


class Tag(Base):
    """New model for ticket tags"""
    __tablename__ = 'tags'

    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False, unique=True)
    color = Column(String(7), nullable=False, default='#808080')  # Hex color
    created_at = Column(DateTime, nullable=False, default=datetime.utcnow)


class TicketTag(Base):
    """Association table for tickets and tags"""
    __tablename__ = 'ticket_tags'

    id = Column(Integer, primary_key=True)
    ticket_id = Column(Integer, ForeignKey('tickets.id', ondelete='CASCADE'), nullable=False)
    tag_id = Column(Integer, ForeignKey('tags.id', ondelete='CASCADE'), nullable=False)
    created_at = Column(DateTime, nullable=False, default=datetime.utcnow)

Command:

alembic revision --autogenerate -m "add tags for tickets"

Generated Migration File (versions/007_add_tags_for_tickets.py):

"""add tags for tickets

Revision ID: 007_tags
Revises: 006_status_migration
Create Date: 2025-01-15 16:00:00.000000
"""

from alembic import op
import sqlalchemy as sa

revision = '007_tags'
down_revision = '006_status_migration'
branch_labels = None
depends_on = None


def upgrade() -> None:
    # Auto-generated - please review!

    # Create tags table
    op.create_table(
        'tags',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('name', sa.String(length=100), nullable=False),
        sa.Column('color', sa.String(length=7), nullable=False),
        sa.Column('created_at', sa.DateTime(), nullable=False),
        sa.PrimaryKeyConstraint('id'),
        sa.UniqueConstraint('name')
    )

    # Create ticket_tags association table
    op.create_table(
        'ticket_tags',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('ticket_id', sa.Integer(), nullable=False),
        sa.Column('tag_id', sa.Integer(), nullable=False),
        sa.Column('created_at', sa.DateTime(), nullable=False),
        sa.ForeignKeyConstraint(['tag_id'], ['tags.id'], ondelete='CASCADE'),
        sa.ForeignKeyConstraint(['ticket_id'], ['tickets.id'], ondelete='CASCADE'),
        sa.PrimaryKeyConstraint('id')
    )

    # Manual additions (not auto-generated)
    op.create_index('ix_tags_name', 'tags', ['name'])
    op.create_index('ix_ticket_tags_ticket_id', 'ticket_tags', ['ticket_id'])
    op.create_index('ix_ticket_tags_tag_id', 'ticket_tags', ['tag_id'])

    # Unique constraint to prevent duplicate tags on same ticket
    op.create_unique_constraint(
        'uq_ticket_tags_ticket_tag',
        'ticket_tags',
        ['ticket_id', 'tag_id']
    )


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

Run the migration:

alembic upgrade head

---

Example 8: Complex Manual Migration

Scenario: Creating a ticket audit log with triggers (manual migration for complex logic).

Command:

alembic revision -m "create ticket audit log with triggers"

Migration File (versions/008_create_audit_log.py):

"""create ticket audit log with triggers

Revision ID: 008_audit_log
Revises: 007_tags
Create Date: 2025-01-15 17:00:00.000000
"""

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql

revision = '008_audit_log'
down_revision = '007_tags'
branch_labels = None
depends_on = None


def upgrade() -> None:
    # Create audit log table
    op.create_table(
        'ticket_audit_log',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('ticket_id', sa.Integer(), nullable=False),
        sa.Column('user_id', sa.Integer(), nullable=True),
        sa.Column('action', sa.String(50), nullable=False),
        sa.Column('field_name', sa.String(100), nullable=True),
        sa.Column('old_value', sa.Text(), nullable=True),
        sa.Column('new_value', sa.Text(), nullable=True),
        sa.Column('changed_at', sa.DateTime(), nullable=False, server_default=sa.text('NOW()')),
        sa.Column('metadata', postgresql.JSONB(), nullable=True),
        sa.ForeignKeyConstraint(['ticket_id'], ['tickets.id'], ondelete='CASCADE'),
        sa.ForeignKeyConstraint(['user_id'], ['users.id'], ondelete='SET NULL')
    )

    # Create indexes
    op.create_index('ix_audit_ticket_id', 'ticket_audit_log', ['ticket_id'])
    op.create_index('ix_audit_changed_at', 'ticket_audit_log', ['changed_at'])
    op.create_index('ix_audit_action', 'ticket_audit_log', ['action'])

    # Create GIN index for JSONB metadata column
    op.execute("""
        CREATE INDEX ix_audit_metadata
        ON ticket_audit_log
        USING gin(metadata)
    """)

    # Create trigger function to automatically log ticket changes
    op.execute("""
        CREATE OR REPLACE FUNCTION log_ticket_changes()
        RETURNS TRIGGER AS $$
        BEGIN
            IF (TG_OP = 'UPDATE') THEN
                -- Log status changes
                IF NEW.status IS DISTINCT FROM OLD.status THEN
                    INSERT INTO ticket_audit_log (ticket_id, action, field_name, old_value, new_value)
                    VALUES (NEW.id, 'status_changed', 'status', OLD.status, NEW.status);
                END IF;

                -- Log priority changes
                IF NEW.priority IS DISTINCT FROM OLD.priority THEN
                    INSERT INTO ticket_audit_log (ticket_id, action, field_name, old_value, new_value)
                    VALUES (NEW.id, 'priority_changed', 'priority', CAST(OLD.priority AS TEXT), CAST(NEW.priority AS TEXT));
                END IF;

                -- Log assignment changes
                IF NEW.assigned_user_id IS DISTINCT FROM OLD.assigned_user_id THEN
                    INSERT INTO ticket_audit_log (ticket_id, action, field_name, old_value, new_value, user_id)
                    VALUES (NEW.id, 'assigned', 'assigned_user_id',
                            CAST(OLD.assigned_user_id AS TEXT),
                            CAST(NEW.assigned_user_id AS TEXT),
                            NEW.assigned_user_id);
                END IF;

            ELSIF (TG_OP = 'INSERT') THEN
                INSERT INTO ticket_audit_log (ticket_id, action, user_id)
                VALUES (NEW.id, 'created', NEW.assigned_user_id);

            ELSIF (TG_OP = 'DELETE') THEN
                INSERT INTO ticket_audit_log (ticket_id, action)
                VALUES (OLD.id, 'deleted');
            END IF;

            RETURN NEW;
        END;
        $$ LANGUAGE plpgsql;
    """)

    # Attach trigger to tickets table
    op.execute("""
        CREATE TRIGGER ticket_changes_trigger
        AFTER INSERT OR UPDATE OR DELETE ON tickets
        FOR EACH ROW
        EXECUTE FUNCTION log_ticket_changes();
    """)


def downgrade() -> None:
    # Drop trigger
    op.execute("DROP TRIGGER IF EXISTS ticket_changes_trigger ON tickets")

    # Drop trigger function
    op.execute("DROP FUNCTION IF EXISTS log_ticket_changes()")

    # Drop table
    op.drop_table('ticket_audit_log')

Run the migration:

alembic upgrade head

---

Example 9: Downgrade Procedures

Scenario: Demonstrating safe downgrade from audit log migration.

Commands:

# Show current revision
alembic current

# Output:
# 008_audit_log (head)

# Show what downgrade -1 will do
alembic downgrade -1 --sql

# Actually downgrade one step
alembic downgrade -1

# Output:
# INFO  [alembic.runtime.migration] Running downgrade 008_audit_log -> 007_tags

# Verify new head
alembic current

# Output:
# 007_tags (head)

# Upgrade back to latest
alembic upgrade head

# Output:
# INFO  [alembic.runtime.migration] Running upgrade 007_tags -> 008_audit_log

# Downgrade to specific revision
alembic downgrade 005_modify_types

# Downgrade all the way to base (empty database)
alembic downgrade base

# Upgrade all the way back
alembic upgrade head

---

Example 10: Creating Migration Branches

Scenario: Creating separate branches for reporting and analytics features.

Commands and Files:

# Create reporting branch from base
alembic revision \
    -m "create reporting branch" \
    --head=base \
    --branch-label=reporting \
    --version-path=alembic/versions/reporting

Generated File (versions/reporting/009_create_reporting_branch.py):

"""create reporting branch

Revision ID: 009_reporting_base
Revises:
Create Date: 2025-01-15 18:00:00.000000
"""

from alembic import op
import sqlalchemy as sa

revision = '009_reporting_base'
down_revision = None
branch_labels = ('reporting',)
depends_on = None


def upgrade() -> None:
    # Create reports table
    op.create_table(
        'reports',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('name', sa.String(200), nullable=False),
        sa.Column('description', sa.Text(), nullable=True),
        sa.Column('report_type', sa.String(50), nullable=False),
        sa.Column('parameters', sa.JSON(), nullable=True),
        sa.Column('created_by', sa.Integer(), nullable=False),
        sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.text('NOW()')),
        sa.ForeignKeyConstraint(['created_by'], ['users.id'], ondelete='CASCADE')
    )


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

# Add another migration to reporting branch
alembic revision \
    -m "add scheduled reports" \
    --head=reporting@head

Generated File (versions/reporting/010_add_scheduled_reports.py):

"""add scheduled reports

Revision ID: 010_scheduled_reports
Revises: 009_reporting_base
Create Date: 2025-01-15 18:30:00.000000
"""

from alembic import op
import sqlalchemy as sa

revision = '010_scheduled_reports'
down_revision = '009_reporting_base'
branch_labels = None
depends_on = None


def upgrade() -> None:
    # Create scheduled reports table
    op.create_table(
        'scheduled_reports',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('report_id', sa.Integer(), nullable=False),
        sa.Column('schedule_cron', sa.String(100), nullable=False),
        sa.Column('recipients', sa.JSON(), nullable=False),
        sa.Column('is_active', sa.Boolean(), nullable=False, server_default='true'),
        sa.Column('last_run_at', sa.DateTime(), nullable=True),
        sa.Column('next_run_at', sa.DateTime(), nullable=True),
        sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.text('NOW()')),
        sa.ForeignKeyConstraint(['report_id'], ['reports.id'], ondelete='CASCADE')
    )


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

Working with branches:

# Show all branches
alembic branches

# Show all heads
alembic heads

# Upgrade specific branch
alembic upgrade reporting@head

# Upgrade all branches
alembic upgrade heads

---

Example 11: Merging Migration Branches

Scenario: Merging reporting branch back into main branch.

Command:

# Merge main and reporting branches
alembic merge \
    -m "merge reporting branch into main" \
    008_audit_log 010_scheduled_reports

Generated File (versions/011_merge_reporting_into_main.py):

"""merge reporting branch into main

Revision ID: 011_merge
Revises: 008_audit_log, 010_scheduled_reports
Create Date: 2025-01-15 19:00:00.000000
"""

from alembic import op
import sqlalchemy as sa

revision = '011_merge'
down_revision = ('008_audit_log', '010_scheduled_reports')
branch_labels = None
depends_on = None


def upgrade() -> None:
    # Usually empty for simple merges
    # Add code here if you need to reconcile conflicts

    # Example: Add a cross-branch constraint
    op.create_foreign_key(
        'fk_reports_created_by',
        'reports',
        'users',
        ['created_by'],
        ['id'],
        ondelete='CASCADE'
    )


def downgrade() -> None:
    # Reverse any changes made in upgrade
    op.drop_constraint('fk_reports_created_by', 'reports', type_='foreignkey')

Apply the merge:

alembic upgrade head

---

Example 12: Online Migration with Minimal Downtime

Scenario: Adding a required column to tickets table without downtime.

Phase 1 - Add Column as Nullable:

alembic revision -m "add resolution notes phase 1 - add column"

"""add resolution notes phase 1 - add column

Revision ID: 012_phase1
Revises: 011_merge
Create Date: 2025-01-16 10:00:00.000000
"""

from alembic import op
import sqlalchemy as sa

revision = '012_phase1'
down_revision = '011_merge'
branch_labels = None
depends_on = None


def upgrade() -> None:
    # Add column as nullable (safe for existing rows)
    op.add_column(
        'tickets',
        sa.Column('resolution_notes', sa.Text(), nullable=True)
    )


def downgrade() -> None:
    op.drop_column('tickets', 'resolution_notes')

Phase 2 - Backfill Data:

alembic revision -m "add resolution notes phase 2 - backfill"

"""add resolution notes phase 2 - backfill

Revision ID: 013_phase2
Revises: 012_phase1
Create Date: 2025-01-16 10:30:00.000000
"""

from alembic import op
import sqlalchemy as sa
from sqlalchemy.sql import table, column

revision = '013_phase2'
down_revision = '012_phase1'
branch_labels = None
depends_on = None


def upgrade() -> None:
    # Backfill resolution_notes for resolved tickets
    # Do this in batches during low-traffic periods

    connection = op.get_bind()
    tickets = table(
        'tickets',
        column('id', sa.Integer),
        column('status', sa.String),
        column('resolution_notes', sa.Text)
    )

    # Set default value for resolved tickets without notes
    connection.execute(
        tickets.update()
        .where(sa.and_(
            tickets.c.status.in_(['RESOLVED', 'CLOSED']),
            tickets.c.resolution_notes.is_(None)
        ))
        .values(resolution_notes='Resolved - details not recorded')
    )


def downgrade() -> None:
    # Clear backfilled data
    connection = op.get_bind()
    tickets = table(
        'tickets',
        column('resolution_notes', sa.Text)
    )

    connection.execute(
        tickets.update()
        .where(tickets.c.resolution_notes == 'Resolved - details not recorded')
        .values(resolution_notes=None)
    )

Phase 3 - Make Column Required:

alembic revision -m "add resolution notes phase 3 - make required"

"""add resolution notes phase 3 - make required

Revision ID: 014_phase3
Revises: 013_phase2
Create Date: 2025-01-16 11:00:00.000000
"""

from alembic import op
import sqlalchemy as sa

revision = '014_phase3'
down_revision = '013_phase2'
branch_labels = None
depends_on = None


def upgrade() -> None:
    # Now that all rows have values, make it non-nullable
    op.alter_column(
        'tickets',
        'resolution_notes',
        nullable=False,
        existing_type=sa.Text(),
        server_default='Pending resolution'
    )


def downgrade() -> None:
    op.alter_column(
        'tickets',
        'resolution_notes',
        nullable=True,
        existing_type=sa.Text(),
        server_default=None
    )

Deployment:

# Deploy phase 1
alembic upgrade 012_phase1

# Wait and monitor

# Deploy phase 2 (can run during low traffic)
alembic upgrade 013_phase2

# Wait and monitor

# Deploy phase 3
alembic upgrade 014_phase3

---

Example 13: Testing Migrations with Pytest

Test File (tests/test_alembic_migrations.py):

"""Tests for Alembic migrations"""

import pytest
from alembic import command
from alembic.config import Config
from sqlalchemy import create_engine, inspect, text
from sqlalchemy.orm import sessionmaker


@pytest.fixture(scope='session')
def alembic_config():
    """Alembic configuration for testing"""
    config = Config("alembic.ini")
    config.set_main_option(
        "sqlalchemy.url",
        "postgresql://localhost/support_test"
    )
    return config


@pytest.fixture
def test_engine(alembic_config):
    """Create test database engine"""
    url = alembic_config.get_main_option("sqlalchemy.url")
    engine = create_engine(url)

    # Create all tables
    command.upgrade(alembic_config, "head")

    yield engine

    # Cleanup
    command.downgrade(alembic_config, "base")
    engine.dispose()


def test_migration_creates_all_tables(test_engine):
    """Test that migrations create all expected tables"""
    inspector = inspect(test_engine)
    tables = inspector.get_table_names()

    expected_tables = [
        'users',
        'customers',
        'tickets',
        'ticket_comments',
        'satisfaction_surveys',
        'tags',
        'ticket_tags',
        'ticket_audit_log',
        'reports',
        'scheduled_reports',
        'alembic_version'
    ]

    for table in expected_tables:
        assert table in tables, f"Table {table} not found in database"


def test_tickets_table_structure(test_engine):
    """Test tickets table has correct columns and types"""
    inspector = inspect(test_engine)
    columns = {col['name']: col for col in inspector.get_columns('tickets')}

    # Check required columns exist
    required_columns = [
        'id', 'customer_id', 'assigned_user_id', 'subject', 'description',
        'status', 'priority', 'created_at', 'updated_at', 'resolved_at',
        'sla_deadline', 'sla_violated', 'first_response_time_seconds',
        'resolution_time_seconds', 'resolution_notes'
    ]

    for col_name in required_columns:
        assert col_name in columns, f"Column {col_name} not found"

    # Check column types
    assert 'integer' in str(columns['id']['type']).lower()
    assert 'varchar' in str(columns['subject']['type']).lower()
    assert 'text' in str(columns['description']['type']).lower()


def test_foreign_keys_exist(test_engine):
    """Test that foreign key constraints are created"""
    inspector = inspect(test_engine)
    fks = inspector.get_foreign_keys('tickets')

    # Should have foreign keys to customers and users
    fk_tables = [fk['referred_table'] for fk in fks]
    assert 'customers' in fk_tables
    assert 'users' in fk_tables


def test_indexes_created(test_engine):
    """Test that performance indexes exist"""
    inspector = inspect(test_engine)
    indexes = inspector.get_indexes('tickets')

    index_names = [idx['name'] for idx in indexes]

    expected_indexes = [
        'ix_tickets_status',
        'ix_tickets_priority',
        'ix_tickets_customer_id',
        'ix_tickets_assigned_user_id',
        'ix_tickets_created_at'
    ]

    for idx_name in expected_indexes:
        assert idx_name in index_names, f"Index {idx_name} not found"


def test_upgrade_downgrade_cycle(alembic_config):
    """Test complete upgrade/downgrade cycle"""
    # Start from base
    command.downgrade(alembic_config, "base")

    # Upgrade to head
    command.upgrade(alembic_config, "head")

    # Downgrade one step
    command.downgrade(alembic_config, "-1")

    # Upgrade back to head
    command.upgrade(alembic_config, "head")


def test_data_persists_after_migration(test_engine, alembic_config):
    """Test that data is preserved during migrations"""
    Session = sessionmaker(bind=test_engine)
    session = Session()

    # Insert test data
    session.execute(text("""
        INSERT INTO customers (email, name, company)
        VALUES ('test@example.com', 'Test Customer', 'Test Corp')
    """))

    session.execute(text("""
        INSERT INTO tickets (customer_id, subject, description, status, priority)
        VALUES (1, 'Test Ticket', 'Test Description', 'OPEN', 'normal')
    """))

    session.commit()

    # Get ticket ID
    result = session.execute(text("SELECT id FROM tickets WHERE subject = 'Test Ticket'"))
    ticket_id = result.scalar()

    session.close()

    # Run a migration (example: downgrade and upgrade)
    command.downgrade(alembic_config, "-1")
    command.upgrade(alembic_config, "head")

    # Verify data still exists
    session = Session()
    result = session.execute(text(f"SELECT subject FROM tickets WHERE id = {ticket_id}"))
    subject = result.scalar()

    assert subject == 'Test Ticket', "Data was lost during migration"
    session.close()


def test_check_constraint_on_satisfaction_rating(test_engine):
    """Test that check constraint prevents invalid ratings"""
    Session = sessionmaker(bind=test_engine)
    session = Session()

    # Insert valid customer and ticket
    session.execute(text("""
        INSERT INTO customers (id, email, name) VALUES (100, 'check@test.com', 'Check Test')
    """))

    session.execute(text("""
        INSERT INTO tickets (id, customer_id, subject, description, status, priority)
        VALUES (100, 100, 'Check Test', 'Test', 'OPEN', 'normal')
    """))

    session.commit()

    # Try to insert invalid rating (should fail)
    with pytest.raises(Exception):
        session.execute(text("""
            INSERT INTO satisfaction_surveys (ticket_id, customer_id, rating, survey_sent_at)
            VALUES (100, 100, 10, NOW())
        """))
        session.commit()

    session.rollback()

    # Insert valid rating (should succeed)
    session.execute(text("""
        INSERT INTO satisfaction_surveys (ticket_id, customer_id, rating, survey_sent_at)
        VALUES (100, 100, 5, NOW())
    """))
    session.commit()
    session.close()


@pytest.mark.slow
def test_migration_performance(alembic_config):
    """Test that full migration completes within time limit"""
    import time

    command.downgrade(alembic_config, "base")

    start = time.time()
    command.upgrade(alembic_config, "head")
    duration = time.time() - start

    # Should complete within 30 seconds
    assert duration < 30, f"Migration took {duration}s, exceeds 30s limit"

Run tests:

# Run all migration tests
pytest tests/test_alembic_migrations.py -v

# Run specific test
pytest tests/test_alembic_migrations.py::test_migration_creates_all_tables -v

# Run with coverage
pytest tests/test_alembic_migrations.py --cov=alembic --cov-report=html

---

Example 14: Rolling Back Failed Migrations

Scenario: A migration fails partway through and needs cleanup.

Simulation:

"""intentionally failing migration

Revision ID: 015_fail_test
Revises: 014_phase3
Create Date: 2025-01-16 12:00:00.000000
"""

from alembic import op
import sqlalchemy as sa

revision = '015_fail_test'
down_revision = '014_phase3'
branch_labels = None
depends_on = None


def upgrade() -> None:
    # This will succeed
    op.create_table(
        'temp_table',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('data', sa.String(100))
    )

    # This will fail (table doesn't exist)
    op.add_column('nonexistent_table', sa.Column('bad_column', sa.Integer()))


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

Recovery Process:

# Attempt migration (will fail)
alembic upgrade head

# Output:
# INFO  [alembic.runtime.migration] Running upgrade 014_phase3 -> 015_fail_test
# ERROR [alembic.runtime.migration] Error running upgrade: Table 'nonexistent_table' does not exist
# FAILED: Target database is not up to date.

# Check current status
alembic current

# Output may show partial application or still at previous revision

# Option 1: Fix the migration and retry
# Edit the migration file to fix the error

# Stamp database to current state (if needed)
alembic stamp 014_phase3

# Try again with fixed migration
alembic upgrade head

# Option 2: Manually clean up and skip the migration
# Connect to database and drop temp_table if it was created
psql $DATABASE_URL -c "DROP TABLE IF EXISTS temp_table"

# Stamp to the failed revision to mark it as applied
alembic stamp 015_fail_test

# Then downgrade it
alembic downgrade -1

# Option 3: Use transaction per migration (recommended)
# Configure in env.py:
context.configure(
    connection=connection,
    target_metadata=target_metadata,
    transaction_per_migration=True  # Each migration in its own transaction
)

# Now failed migrations automatically rollback

---

Example 15: Production Deployment Workflow

Deployment Script (scripts/deploy_migrations.sh):

#!/bin/bash

# Production migration deployment script
# Usage: ./scripts/deploy_migrations.sh

set -e  # Exit on any error
set -u  # Exit on undefined variable

echo "======================================"
echo "Production Migration Deployment"
echo "======================================"

# Configuration
BACKUP_DIR="${BACKUP_DIR:-./backups}"
TIMESTAMP=$(date +%Y%m%d_%H%M%S)
BACKUP_FILE="${BACKUP_DIR}/pre_migration_${TIMESTAMP}.sql"
LOG_FILE="./logs/migration_${TIMESTAMP}.log"

# Ensure directories exist
mkdir -p "$BACKUP_DIR"
mkdir -p "./logs"

# Functions
log() {
    echo "[$(date +'%Y-%m-%d %H:%M:%S')] $*" | tee -a "$LOG_FILE"
}

error_exit() {
    log "ERROR: $1"
    exit 1
}

# Validate environment
log "Validating environment..."
if [ -z "${DATABASE_URL:-}" ]; then
    error_exit "DATABASE_URL environment variable not set"
fi

if [ -z "${DB_PASSWORD:-}" ]; then
    error_exit "DB_PASSWORD environment variable not set"
fi

# Check Alembic is installed
if ! command -v alembic &> /dev/null; then
    error_exit "Alembic not found. Please install: pip install alembic"
fi

# Step 1: Backup database
log "Creating database backup..."
pg_dump "$DATABASE_URL" > "$BACKUP_FILE" || error_exit "Backup failed"
log "Backup created: $BACKUP_FILE"

# Step 2: Show current status
log "Current migration status:"
alembic current 2>&1 | tee -a "$LOG_FILE"

# Step 3: Show pending migrations
log "Checking for pending migrations..."
CURRENT_REV=$(alembic current | grep -oP 'Rev: \K\w+' || echo "base")
HEAD_REV=$(alembic heads | awk '{print $1}')

if [ "$CURRENT_REV" == "$HEAD_REV" ]; then
    log "Database is already up to date. No migrations needed."
    exit 0
fi

log "Pending migrations will be applied from $CURRENT_REV to $HEAD_REV"

# Step 4: Confirm with user
read -p "Proceed with migration? (yes/no): " -r
if [[ ! $REPLY =~ ^[Yy][Ee][Ss]$ ]]; then
    log "Migration cancelled by user"
    exit 0
fi

# Step 5: Run migrations with timeout
log "Running migrations..."
timeout 300 alembic upgrade head 2>&1 | tee -a "$LOG_FILE" || {
    log "Migration failed or timed out!"
    log "Attempting to restore from backup..."

    # Restore from backup
    psql "$DATABASE_URL" < "$BACKUP_FILE" || error_exit "Restore failed!"

    log "Database restored from backup"
    error_exit "Migration failed. Database restored to previous state."
}

# Step 6: Verify migration success
log "Verifying migration status..."
NEW_REV=$(alembic current | grep -oP 'Rev: \K\w+' || echo "none")

if [ "$NEW_REV" != "$HEAD_REV" ]; then
    log "WARNING: Migration incomplete. Current: $NEW_REV, Expected: $HEAD_REV"
    log "Restoring from backup..."

    psql "$DATABASE_URL" < "$BACKUP_FILE" || error_exit "Restore failed!"

    error_exit "Migration verification failed. Database restored."
fi

# Step 7: Run post-migration checks
log "Running post-migration checks..."

# Check database connectivity
psql "$DATABASE_URL" -c "SELECT 1" > /dev/null || error_exit "Database connectivity check failed"

# Check critical tables exist
CRITICAL_TABLES=("users" "customers" "tickets")
for table in "${CRITICAL_TABLES[@]}"; do
    COUNT=$(psql "$DATABASE_URL" -t -c "SELECT COUNT(*) FROM information_schema.tables WHERE table_name='$table'")
    if [ "$COUNT" -eq 0 ]; then
        error_exit "Critical table $table not found!"
    fi
done

log "Post-migration checks passed"

# Step 8: Cleanup old backups (keep last 10)
log "Cleaning up old backups..."
ls -t "$BACKUP_DIR"/*.sql 2>/dev/null | tail -n +11 | xargs -r rm
log "Old backups cleaned up (kept last 10)"

# Step 9: Final summary
log "======================================"
log "Migration completed successfully!"
log "Previous revision: $CURRENT_REV"
log "Current revision: $NEW_REV"
log "Backup location: $BACKUP_FILE"
log "Log location: $LOG_FILE"
log "======================================"

exit 0

Usage:

# Set environment variables
export DATABASE_URL="postgresql://user:pass@localhost/support_prod"
export DB_PASSWORD="secure_password"

# Make script executable
chmod +x scripts/deploy_migrations.sh

# Run deployment
./scripts/deploy_migrations.sh

# Output:
# ======================================
# Production Migration Deployment
# ======================================
# [2025-01-16 12:00:00] Validating environment...
# [2025-01-16 12:00:01] Creating database backup...
# [2025-01-16 12:00:15] Backup created: ./backups/pre_migration_20250116_120000.sql
# [2025-01-16 12:00:15] Current migration status:
# Rev: 014_phase3 (head)
# [2025-01-16 12:00:16] Checking for pending migrations...
# Proceed with migration? (yes/no): yes
# [2025-01-16 12:00:20] Running migrations...
# INFO  [alembic.runtime.migration] Running upgrade 014_phase3 -> 015_new_feature
# [2025-01-16 12:00:25] Verifying migration status...
# [2025-01-16 12:00:26] Running post-migration checks...
# [2025-01-16 12:00:27] Post-migration checks passed
# [2025-01-16 12:00:27] Cleaning up old backups...
# ======================================
# Migration completed successfully!
# Previous revision: 014_phase3
# Current revision: 015_new_feature
# Backup location: ./backups/pre_migration_20250116_120000.sql
# Log location: ./logs/migration_20250116_120000.log
# ======================================

---

Example 16: Batch Data Migration

Scenario: Computing and backfilling metrics for large ticket table.

Command:

alembic revision -m "compute and backfill ticket metrics"

Migration File (versions/016_compute_ticket_metrics.py):

"""compute and backfill ticket metrics

Revision ID: 016_metrics
Revises: 015_new_feature
Create Date: 2025-01-16 13:00:00.000000
"""

from alembic import op
import sqlalchemy as sa
from sqlalchemy.sql import table, column, select
import time

revision = '016_metrics'
down_revision = '015_new_feature'
branch_labels = None
depends_on = None


def upgrade() -> None:
    # Add metrics columns
    op.add_column('tickets',
        sa.Column('total_comments', sa.Integer(), nullable=True, server_default='0'))

    op.add_column('tickets',
        sa.Column('customer_responses', sa.Integer(), nullable=True, server_default='0'))

    op.add_column('tickets',
        sa.Column('agent_responses', sa.Integer(), nullable=True, server_default='0'))

    # Compute metrics in batches
    connection = op.get_bind()

    # Get total number of tickets
    result = connection.execute(sa.text("SELECT COUNT(*) FROM tickets"))
    total_tickets = result.scalar()

    print(f"Processing {total_tickets} tickets in batches...")

    batch_size = 1000
    processed = 0

    while processed < total_tickets:
        # Get batch of ticket IDs
        batch_ids = connection.execute(
            sa.text(f"""
                SELECT id FROM tickets
                ORDER BY id
                LIMIT {batch_size} OFFSET {processed}
            """)
        ).fetchall()

        if not batch_ids:
            break

        ticket_ids = [row[0] for row in batch_ids]

        # Compute metrics for this batch
        for ticket_id in ticket_ids:
            # Count total comments
            total = connection.execute(
                sa.text(f"""
                    SELECT COUNT(*) FROM ticket_comments
                    WHERE ticket_id = {ticket_id}
                """)
            ).scalar()

            # Count customer responses
            customer_count = connection.execute(
                sa.text(f"""
                    SELECT COUNT(*) FROM ticket_comments
                    WHERE ticket_id = {ticket_id}
                    AND customer_id IS NOT NULL
                """)
            ).scalar()

            # Count agent responses
            agent_count = connection.execute(
                sa.text(f"""
                    SELECT COUNT(*) FROM ticket_comments
                    WHERE ticket_id = {ticket_id}
                    AND user_id IS NOT NULL
                """)
            ).scalar()

            # Update ticket metrics
            connection.execute(
                sa.text(f"""
                    UPDATE tickets
                    SET total_comments = {total},
                        customer_responses = {customer_count},
                        agent_responses = {agent_count}
                    WHERE id = {ticket_id}
                """)
            )

        processed += len(ticket_ids)
        progress = (processed / total_tickets) * 100
        print(f"Processed {processed}/{total_tickets} tickets ({progress:.1f}%)")

        # Small delay to reduce database load
        time.sleep(0.1)

    # Make columns non-nullable
    op.alter_column('tickets', 'total_comments', nullable=False)
    op.alter_column('tickets', 'customer_responses', nullable=False)
    op.alter_column('tickets', 'agent_responses', nullable=False)

    # Create indexes for metrics
    op.create_index('ix_tickets_total_comments', 'tickets', ['total_comments'])


def downgrade() -> None:
    op.drop_index('ix_tickets_total_comments', 'tickets')
    op.drop_column('tickets', 'agent_responses')
    op.drop_column('tickets', 'customer_responses')
    op.drop_column('tickets', 'total_comments')

---

Example 17: Adding Enums and Constraints

Scenario: Adding ticket category enum and related constraints.

Command:

alembic revision -m "add ticket categories with constraints"

Migration File (versions/017_add_ticket_categories.py):

"""add ticket categories with constraints

Revision ID: 017_categories
Revises: 016_metrics
Create Date: 2025-01-16 14:00:00.000000
"""

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql

revision = '017_categories'
down_revision = '016_metrics'
branch_labels = None
depends_on = None


def upgrade() -> None:
    # Create category enum type
    category_enum = postgresql.ENUM(
        'billing',
        'technical',
        'feature_request',
        'bug_report',
        'general_inquiry',
        'account_management',
        name='ticket_category_enum',
        create_type=True
    )
    category_enum.create(op.get_bind())

    # Add category column
    op.add_column('tickets',
        sa.Column('category', category_enum, nullable=True))

    # Set default category based on existing data
    op.execute("""
        UPDATE tickets
        SET category = 'general_inquiry'::ticket_category_enum
        WHERE category IS NULL
    """)

    # Make category required
    op.alter_column('tickets', 'category', nullable=False)

    # Create subcategory table
    op.create_table(
        'ticket_categories',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('category', category_enum, nullable=False),
        sa.Column('subcategory', sa.String(100), nullable=False),
        sa.Column('description', sa.Text(), nullable=True),
        sa.Column('sla_hours', sa.Integer(), nullable=False),
        sa.Column('is_active', sa.Boolean(), nullable=False, server_default='true'),
        sa.UniqueConstraint('category', 'subcategory', name='uq_category_subcategory')
    )

    # Add subcategory to tickets
    op.add_column('tickets',
        sa.Column('subcategory_id', sa.Integer(), nullable=True))

    op.create_foreign_key(
        'fk_tickets_subcategory',
        'tickets',
        'ticket_categories',
        ['subcategory_id'],
        ['id'],
        ondelete='SET NULL'
    )

    # Insert default subcategories
    op.execute("""
        INSERT INTO ticket_categories (category, subcategory, sla_hours) VALUES
        ('billing', 'Invoice Question', 24),
        ('billing', 'Payment Issue', 12),
        ('billing', 'Refund Request', 48),
        ('technical', 'Login Problem', 4),
        ('technical', 'Performance Issue', 8),
        ('technical', 'Integration Problem', 24),
        ('feature_request', 'New Feature', 168),
        ('feature_request', 'Enhancement', 168),
        ('bug_report', 'Critical Bug', 4),
        ('bug_report', 'Minor Bug', 48),
        ('general_inquiry', 'How To', 24),
        ('general_inquiry', 'Information Request', 24),
        ('account_management', 'Update Details', 24),
        ('account_management', 'Close Account', 48)
    """)

    # Create indexes
    op.create_index('ix_tickets_category', 'tickets', ['category'])
    op.create_index('ix_tickets_subcategory_id', 'tickets', ['subcategory_id'])


def downgrade() -> None:
    # Drop indexes
    op.drop_index('ix_tickets_subcategory_id', 'tickets')
    op.drop_index('ix_tickets_category', 'tickets')

    # Drop foreign key and column
    op.drop_constraint('fk_tickets_subcategory', 'tickets', type_='foreignkey')
    op.drop_column('tickets', 'subcategory_id')

    # Drop subcategory table
    op.drop_table('ticket_categories')

    # Drop category column
    op.drop_column('tickets', 'category')

    # Drop enum type
    category_enum = postgresql.ENUM(name='ticket_category_enum')
    category_enum.drop(op.get_bind())

---

Example 18: Multi-Table Data Migration

Scenario: Restructuring customer contact information into separate table.

Command:

alembic revision -m "extract customer contacts to separate table"

Migration File (versions/018_extract_customer_contacts.py):

"""extract customer contacts to separate table

Revision ID: 018_contacts
Revises: 017_categories
Create Date: 2025-01-16 15:00:00.000000
"""

from alembic import op
import sqlalchemy as sa
from sqlalchemy.sql import table, column

revision = '018_contacts'
down_revision = '017_categories'
branch_labels = None
depends_on = None


def upgrade() -> None:
    # Create new customer_contacts table
    op.create_table(
        'customer_contacts',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('customer_id', sa.Integer(), nullable=False),
        sa.Column('contact_type', sa.String(50), nullable=False),
        sa.Column('contact_value', sa.String(255), nullable=False),
        sa.Column('is_primary', sa.Boolean(), nullable=False, server_default='false'),
        sa.Column('is_verified', sa.Boolean(), nullable=False, server_default='false'),
        sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.text('NOW()')),
        sa.ForeignKeyConstraint(['customer_id'], ['customers.id'], ondelete='CASCADE'),
        sa.CheckConstraint(
            "contact_type IN ('email', 'phone', 'mobile')",
            name='ck_contacts_type'
        )
    )

    op.create_index('ix_contacts_customer_id', 'customer_contacts', ['customer_id'])
    op.create_index('ix_contacts_type', 'customer_contacts', ['contact_type'])
    op.create_index('ix_contacts_primary', 'customer_contacts', ['customer_id', 'is_primary'])

    # Migrate existing data from customers table
    connection = op.get_bind()

    # Migrate email addresses
    connection.execute(sa.text("""
        INSERT INTO customer_contacts (customer_id, contact_type, contact_value, is_primary, is_verified)
        SELECT id, 'email', email, true, true
        FROM customers
        WHERE email IS NOT NULL AND email != ''
    """))

    # Migrate phone numbers
    connection.execute(sa.text("""
        INSERT INTO customer_contacts (customer_id, contact_type, contact_value, is_primary)
        SELECT id, 'phone', phone, false
        FROM customers
        WHERE phone IS NOT NULL AND phone != ''
    """))

    # Add primary_contact_id to customers table
    op.add_column('customers',
        sa.Column('primary_contact_id', sa.Integer(), nullable=True))

    # Set primary_contact_id to the email contact for each customer
    connection.execute(sa.text("""
        UPDATE customers c
        SET primary_contact_id = cc.id
        FROM customer_contacts cc
        WHERE cc.customer_id = c.id
        AND cc.contact_type = 'email'
        AND cc.is_primary = true
    """))

    # Create foreign key
    op.create_foreign_key(
        'fk_customers_primary_contact',
        'customers',
        'customer_contacts',
        ['primary_contact_id'],
        ['id'],
        ondelete='SET NULL'
    )

    # Now we can drop the old columns (optional - keep for backwards compatibility)
    # op.drop_column('customers', 'phone')
    # We keep email column for now as it's heavily used


def downgrade() -> None:
    # Drop foreign key
    op.drop_constraint('fk_customers_primary_contact', 'customers', type_='foreignkey')

    # Drop primary_contact_id column
    op.drop_column('customers', 'primary_contact_id')

    # Restore phone data from contacts table (if we dropped it)
    # connection = op.get_bind()
    # connection.execute(sa.text("""
    #     UPDATE customers c
    #     SET phone = cc.contact_value
    #     FROM customer_contacts cc
    #     WHERE cc.customer_id = c.id
    #     AND cc.contact_type = 'phone'
    #     AND cc.is_primary = true
    # """))

    # Drop contacts table
    op.drop_table('customer_contacts')

---

Summary

These 18 examples demonstrate:

1. Initial Setup: Creating foundational schema 2. Schema Evolution: Adding columns, tables, and indexes 3. Data Migrations: Transforming and migrating existing data 4. Autogenerate: Using SQLAlchemy models to generate migrations 5. Complex Migrations: Manual migrations with triggers and functions 6. Branching: Managing parallel development streams 7. Zero-Downtime: Multi-phase migrations for production 8. Testing: Comprehensive test coverage for migrations 9. Production Deployment: Safe deployment workflows with backups 10. Performance: Batch processing for large datasets

Each example is production-ready and can be adapted for your specific customer support system needs. Always review and test migrations thoroughly before applying them to production databases.

Alembic Database Migrations for Customer Support Systems

Overview

This skill provides comprehensive guidance for managing database schema evolution using Alembic in customer support environments. Whether you're building a ticketing system, managing customer data, or maintaining complex support infrastructure, Alembic helps you safely evolve your database schema while preserving data integrity.

What is Alembic?

Alembic is a lightweight database migration tool for SQLAlchemy that provides:

• Version Control for Database Schemas: Track every change to your database structure
• Automated Migration Generation: Detect schema differences automatically
• Safe Rollback Capabilities: Revert changes when issues arise
• Team Collaboration: Merge schema changes from multiple developers
• Production-Ready Workflows: Deploy schema changes with confidence

For customer support teams, this means you can:

• Add new features without database downtime
• Safely modify ticket tracking schemas
• Migrate data as business requirements evolve
• Maintain consistency across dev, staging, and production environments

Quick Start

Installation

# Install Alembic with PostgreSQL support
pip install alembic sqlalchemy psycopg2-binary

# Or add to your requirements.txt
echo "alembic>=1.13.0" >> requirements.txt
echo "sqlalchemy>=2.0.0" >> requirements.txt
echo "psycopg2-binary>=2.9.0" >> requirements.txt
pip install -r requirements.txt

Initialize Your Project

# Initialize Alembic in your project
alembic init alembic

# This creates:
# - alembic/                 Directory for migrations
# - alembic/versions/        Individual migration files
# - alembic/env.py          Environment configuration
# - alembic.ini             Main configuration file

Configure Database Connection

Edit alembic.ini to set your database URL:

# For development
sqlalchemy.url = postgresql://user:password@localhost/support_db

# For production, use environment variables (see below)

Better practice - use environment variables in alembic/env.py:

import os
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context

# Import your SQLAlchemy models
from myapp.models import Base

config = context.config

# Override database URL from environment
database_url = os.getenv(
    'DATABASE_URL',
    'postgresql://localhost/support_dev'
)
config.set_main_option('sqlalchemy.url', database_url)

# Set target metadata for autogenerate
target_metadata = Base.metadata

Create Your First Migration

Option 1: Manual Migration

# Create empty migration file
alembic revision -m "create initial support tables"

This generates a file like alembic/versions/abc123_create_initial_support_tables.py:

"""create initial support tables

Revision ID: abc123
Revises:
Create Date: 2025-01-15 10:00:00.000000
"""

from alembic import op
import sqlalchemy as sa

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

def upgrade() -> None:
    # Create tickets table
    op.create_table(
        'tickets',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('title', sa.String(200), nullable=False),
        sa.Column('description', sa.Text(), nullable=True),
        sa.Column('status', sa.String(50), nullable=False, server_default='open'),
        sa.Column('priority', sa.String(20), nullable=False, server_default='normal'),
        sa.Column('created_at', sa.DateTime(), nullable=False),
        sa.Column('updated_at', sa.DateTime(), nullable=False),
    )

    # Create indexes for common queries
    op.create_index('ix_tickets_status', 'tickets', ['status'])
    op.create_index('ix_tickets_created_at', 'tickets', ['created_at'])

def downgrade() -> None:
    op.drop_index('ix_tickets_created_at', 'tickets')
    op.drop_index('ix_tickets_status', 'tickets')
    op.drop_table('tickets')

Option 2: Autogenerate Migration

First, define your models using SQLAlchemy:

# myapp/models.py
from sqlalchemy import Column, Integer, String, Text, DateTime
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime

Base = declarative_base()

class Ticket(Base):
    __tablename__ = 'tickets'

    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    description = Column(Text)
    status = Column(String(50), nullable=False, default='open')
    priority = Column(String(20), nullable=False, default='normal')
    created_at = Column(DateTime, nullable=False, default=datetime.utcnow)
    updated_at = Column(DateTime, nullable=False, default=datetime.utcnow, onupdate=datetime.utcnow)

Then autogenerate the migration:

# Alembic compares your models to the database and generates migration
alembic revision --autogenerate -m "create initial support tables"

Important: Always review autogenerated migrations before running them!

Apply Migrations

# Apply all pending migrations to database
alembic upgrade head

# You should see output like:
# INFO  [alembic.runtime.migration] Running upgrade  -> abc123, create initial support tables

Check Migration Status

# Show current database revision
alembic current

# Show migration history
alembic history

# Show detailed current status
alembic current --verbose

Key Features for Support Teams

1. Safe Schema Evolution

Modify your support system database without downtime or data loss:

# Add a new column to track customer satisfaction
alembic revision -m "add satisfaction rating to tickets"

def upgrade() -> None:
    op.add_column('tickets',
        sa.Column('satisfaction_rating', sa.Integer(), nullable=True)
    )

def downgrade() -> None:
    op.drop_column('tickets', 'satisfaction_rating')

2. Data Migrations

Transform existing data during schema changes:

"""convert ticket priorities to new system

Revision ID: def456
Revises: abc123
"""

from alembic import op
from sqlalchemy.sql import table, column

def upgrade() -> None:
    # Map old priority values to new ones
    tickets = table('tickets', column('priority', sa.String))

    connection = op.get_bind()
    connection.execute(
        tickets.update().where(
            tickets.c.priority == 'high'
        ).values(priority='urgent')
    )

def downgrade() -> None:
    # Reverse the mapping
    tickets = table('tickets', column('priority', sa.String))

    connection = op.get_bind()
    connection.execute(
        tickets.update().where(
            tickets.c.priority == 'urgent'
        ).values(priority='high')
    )

3. Rollback Capabilities

If something goes wrong, easily revert:

# Rollback last migration
alembic downgrade -1

# Rollback to specific revision
alembic downgrade abc123

# Rollback to empty database
alembic downgrade base

4. Branch Management

Handle parallel development from multiple teams:

# Create feature branch for reporting module
alembic revision -m "reporting branch" \
    --branch-label=reporting \
    --head=base

# Create migration on specific branch
alembic revision -m "add report tables" \
    --head=reporting@head

# Merge branches when ready
alembic merge -m "merge reporting into main" \
    main@head reporting@head

5. Testing Migrations

Ensure migrations work before production:

# tests/test_migrations.py
import pytest
from alembic import command
from alembic.config import Config

def test_migration_upgrade_downgrade():
    config = Config("alembic.ini")
    config.set_main_option("sqlalchemy.url", "postgresql://localhost/test_db")

    # Test upgrade
    command.upgrade(config, "head")

    # Test downgrade
    command.downgrade(config, "base")

    # Test upgrade again
    command.upgrade(config, "head")

Common Customer Support Use Cases

Use Case 1: Adding User Assignment to Tickets

alembic revision -m "add user assignment to tickets"

def upgrade() -> None:
    # Add user_id column
    op.add_column('tickets',
        sa.Column('assigned_user_id', sa.Integer(), nullable=True)
    )

    # Create foreign key to users table
    op.create_foreign_key(
        'fk_tickets_assigned_user',
        'tickets', 'users',
        ['assigned_user_id'], ['id'],
        ondelete='SET NULL'
    )

    # Add index for performance
    op.create_index(
        'ix_tickets_assigned_user_id',
        'tickets',
        ['assigned_user_id']
    )

def downgrade() -> None:
    op.drop_index('ix_tickets_assigned_user_id', 'tickets')
    op.drop_constraint('fk_tickets_assigned_user', 'tickets', type_='foreignkey')
    op.drop_column('tickets', 'assigned_user_id')

Use Case 2: Tracking Ticket Resolution Time

alembic revision --autogenerate -m "add resolution tracking"

def upgrade() -> None:
    # Add resolved_at timestamp
    op.add_column('tickets',
        sa.Column('resolved_at', sa.DateTime(), nullable=True)
    )

    # Add computed resolution time in seconds
    op.add_column('tickets',
        sa.Column('resolution_time_seconds', sa.Integer(), nullable=True)
    )

def downgrade() -> None:
    op.drop_column('tickets', 'resolution_time_seconds')
    op.drop_column('tickets', 'resolved_at')

Use Case 3: Customer Satisfaction Survey

alembic revision -m "create satisfaction survey table"

def upgrade() -> None:
    op.create_table(
        'satisfaction_surveys',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('ticket_id', sa.Integer(), nullable=False),
        sa.Column('rating', sa.Integer(), nullable=False),
        sa.Column('feedback', sa.Text(), nullable=True),
        sa.Column('created_at', sa.DateTime(), nullable=False),
        sa.ForeignKeyConstraint(
            ['ticket_id'], ['tickets.id'],
            ondelete='CASCADE'
        )
    )

    op.create_index('ix_surveys_ticket_id', 'satisfaction_surveys', ['ticket_id'])
    op.create_index('ix_surveys_rating', 'satisfaction_surveys', ['rating'])

def downgrade() -> None:
    op.drop_index('ix_surveys_rating', 'satisfaction_surveys')
    op.drop_index('ix_surveys_ticket_id', 'satisfaction_surveys')
    op.drop_table('satisfaction_surveys')

Migration Workflow Best Practices

Development Workflow

1. Make model changes in your SQLAlchemy models 2. Generate migration: alembic revision --autogenerate -m "description" 3. Review migration file - autogenerate isn't perfect! 4. Test locally: alembic upgrade head 5. Test downgrade: alembic downgrade -1 6. Commit migration file to version control

Staging Workflow

1. Deploy code to staging environment 2. Backup staging database 3. Run migrations: alembic upgrade head 4. Test application thoroughly 5. Test rollback if time permits: alembic downgrade -1 then alembic upgrade head

Production Workflow

1. Schedule maintenance window (if needed) 2. Create production backup 3. Deploy code to production 4. Run migrations: alembic upgrade head 5. Monitor application for issues 6. Keep rollback plan ready

Emergency Rollback

# If migration causes issues in production
alembic downgrade -1

# Or downgrade to specific known-good revision
alembic downgrade abc123

# Then deploy previous code version

Troubleshooting

Problem: "Multiple heads exist"

Cause: You have divergent migration branches that need merging.

Solution:

# Show all heads
alembic heads

# Merge them
alembic merge heads -m "merge migration branches"

# Apply the merge
alembic upgrade head

Problem: "Can't locate revision identified by 'xyz'"

Cause: Migration file missing or database revision table corrupted.

Solution:

# Check current database state
alembic current

# Check migration history
alembic history

# If needed, manually stamp database to correct revision
alembic stamp head  # or specific revision

Problem: Migration fails partway through

Cause: SQL error, constraint violation, or data issue.

Solution:

# 1. Check current state
alembic current

# 2. Fix the underlying issue (database constraint, data problem, etc.)

# 3. Try migration again
alembic upgrade head

# 4. If migration script needs fixing:
#    - Edit the migration file
#    - Stamp to previous revision
#    - Run migration again
alembic stamp previous_revision
alembic upgrade head

Problem: Autogenerate creates too many/wrong changes

Cause: Difference in type comparison or server defaults.

Solution: Configure env.py to filter or customize autogenerate:

def run_migrations_online():
    context.configure(
        connection=connection,
        target_metadata=target_metadata,
        compare_type=True,
        compare_server_default=True,
        include_object=include_object,
        include_name=include_name
    )

def include_object(object, name, type_, reflected, compare_to):
    """Filter out test tables and temp tables"""
    if type_ == "table" and (name.startswith("test_") or name.startswith("temp_")):
        return False
    return True

def include_name(name, type_, parent_names):
    """Filter schemas"""
    if type_ == "schema" and name in ("information_schema", "pg_catalog"):
        return False
    return True

File Size and Structure Reference

This skill package includes:

1. SKILL.md (20KB+): Comprehensive skill definition with all migration scenarios 2. README.md (This file, 10KB+): Quick start and overview 3. EXAMPLES.md (15KB+): 15+ practical, runnable examples

Additional Resources

• Official Documentation: https://alembic.sqlalchemy.org/
• SQLAlchemy Documentation: https://docs.sqlalchemy.org/
• PostgreSQL Documentation: https://www.postgresql.org/docs/

Getting Help

Common questions:

Q: Should I use manual or autogenerate for migrations? A: Use autogenerate for simple schema changes, but always review the generated code. Use manual migrations for complex data transformations or when you need precise control.

Q: How do I handle large data migrations? A: Process data in batches to avoid memory issues and reduce database lock time. See the data migration examples in EXAMPLES.md.

Q: Can I run migrations in production without downtime? A: Yes, using multi-phase migrations. Add new columns as nullable, populate data in background, then make required. See zero-downtime migrations in SKILL.md.

Q: How do I test migrations? A: Write tests that apply migrations to a test database, verify schema changes, and test upgrade/downgrade cycles. See testing section in SKILL.md.

Q: What if multiple developers create migrations at the same time? A: Alembic will create multiple heads. Merge them using alembic merge heads. Consider using branch labels for team-specific work.

Next Steps

1. Review SKILL.md for comprehensive documentation 2. Check EXAMPLES.md for practical, copy-paste examples 3. Set up your first migration following the Quick Start above 4. Practice upgrade/downgrade cycles in development 5. Implement CI/CD checks for migrations 6. Establish team migration guidelines

Support

For customer support specific questions about this skill package, consult your team lead or check your internal documentation.

For Alembic-specific issues:

• Check the official documentation
• Search GitHub issues: https://github.com/sqlalchemy/alembic/issues
• Ask on Stack Overflow with the alembic tag