Taming Database Evolution with Alembic¶

Objective: Master Alembic migrations to manage database schema changes with surgical precision. When your database schema evolves faster than your application, when your team needs reproducible database changes, when production deployments require zero-downtime schema updates—Alembic becomes your weapon of choice.

Database migrations are the bridge between your evolving application and its persistent storage. Without them, you're flying blind into production with schema changes that could break everything. This guide shows you how to wield Alembic with the precision of a seasoned backend engineer.

0) Prerequisites (Read Once, Live by Them)¶

The Five Commandments¶

Understand the migration lifecycle
Schema evolution and version control
Team workflows and collaboration
Production deployment safety
Master the Alembic workflow
Autogenerate vs manual migrations
Upgrade and downgrade operations
Environment-specific configurations
Know your database
SQLAlchemy model definitions
Database-specific features and limitations
Performance implications of schema changes
Validate everything
Review autogenerated migrations
Test upgrade and downgrade paths
Verify data integrity after migrations
Plan for production
Zero-downtime deployments
Rollback strategies
CI/CD integration

Why These Principles: Database migrations are the foundation of reliable application deployment. Understanding Alembic's workflow, limitations, and best practices is essential for maintaining data integrity across environments.

1) Prerequisites (The Foundation)¶

System Requirements¶

# Python 3.11+ (required for modern async features)
python --version

# Core dependencies
pip install alembic sqlalchemy psycopg2-binary

# Optional but recommended
pip install alembic[postgresql]  # PostgreSQL-specific features
pip install alembic[mysql]       # MySQL-specific features
pip install alembic[sqlite]      # SQLite-specific features

Database Setup¶

# PostgreSQL (recommended for production)
createdb myapp_dev
createdb myapp_test
createdb myapp_prod

# MySQL (alternative)
mysql -u root -p -e "CREATE DATABASE myapp_dev;"
mysql -u root -p -e "CREATE DATABASE myapp_test;"
mysql -u root -p -e "CREATE DATABASE myapp_prod;"

# SQLite (development only)
touch myapp.db

Why These Dependencies: Alembic requires SQLAlchemy for model introspection and database-specific drivers for connectivity. PostgreSQL is recommended for production due to its advanced features and reliability.

2) Initial Setup (The Foundation)¶

Project Structure¶

myapp/
├── alembic.ini
├── migrations/
│   ├── env.py
│   ├── script.py.mako
│   └── versions/
├── models.py
├── config.py
└── requirements.txt

SQLAlchemy Models¶

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

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    full_name = Column(String(100), nullable=True)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    # Relationships
    posts = relationship("Post", back_populates="author")

class Post(Base):
    __tablename__ = 'posts'

    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(Text, nullable=True)
    author_id = Column(Integer, ForeignKey('users.id'), nullable=False)
    is_published = Column(Boolean, default=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    # Relationships
    author = relationship("User", back_populates="posts")

Initialize Alembic¶

# Initialize Alembic in your project
alembic init migrations

# This creates:
# - alembic.ini (configuration file)
# - migrations/ directory with:
#   - env.py (environment configuration)
#   - script.py.mako (migration template)
#   - versions/ (migration files)

Why This Structure: Alembic requires a specific directory structure to manage migrations. The migrations/ directory contains all migration-related files, while alembic.ini provides global configuration.

3) Configuring env.py (The Heart)¶

Environment Configuration¶

# migrations/env.py
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context
from myapp.models import Base

# Import your models here to ensure they're registered
from myapp.models import User, Post

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config

# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# add your model's MetaData object here
# for 'autogenerate' support
target_metadata = Base.metadata

# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.

def run_migrations_offline():
    """Run migrations in 'offline' mode.

    This configures the context with just a URL
    and not an Engine, though an Engine is acceptable
    here as well.  By skipping the Engine creation
    we don't even need a DBAPI to be available.

    Calls to context.execute() here emit the given string to the
    script output.

    """
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        compare_type=True,
        compare_server_default=True,
    )

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

def run_migrations_online():
    """Run migrations in 'online' mode.

    In this scenario we need to create an Engine
    and associate a connection with the context.

    """
    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,
            compare_type=True,
            compare_server_default=True,
        )

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

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

Alembic Configuration¶

# alembic.ini
[alembic]
# path to migration scripts
script_location = migrations

# template used to generate migration file names; The default value is %%(rev)s_%%(slug)s
# Uncomment the line below if you want the files to be prepended with date and time
# file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s

# sys.path path, will be prepended to sys.path if present.
# defaults to the current working directory.
prepend_sys_path = .

# timezone to use when rendering the date within the migration file
# as well as the filename.
# If specified, requires the python-dateutil library that can be
# installed by adding `alembic[tz]` to the pip requirements
# string value is passed to dateutil.tz.gettz()
# leave blank for localtime
# timezone =

# max length of characters to apply to the
# "slug" field
# truncate_slug_length = 40

# set to 'true' to run the environment during
# the 'revision' command, regardless of autogenerate
# revision_environment = false

# set to 'true' to allow .pyc and .pyo files without
# a source .py file to be detected as revisions in the
# versions/ directory
# sourceless = false

# version number format (use SHA-based IDs instead)
version_num_format = %s

# version path separator; As mentioned above, this is the character used to split
# version_locations. The default within new alembic.ini files is "os", which uses
# os.pathsep. If this key is omitted entirely, it falls back to the legacy
# behavior of splitting on spaces and/or commas.
# Valid values for version_path_separator are:
#
# version_path_separator = :
# version_path_separator = ;
# version_path_separator = space
version_path_separator = os

# set to 'true' to search source files recursively
# in each "version_locations" directory
# new in Alembic version 1.10
# recursive_version_locations = false

# the output encoding used when revision files
# are written from script.py.mako
# output_encoding = utf-8

sqlalchemy.url = postgresql://user:password@localhost/myapp_dev

[post_write_hooks]
# post_write_hooks defines scripts or Python functions to be run
# immediately after a new revision file is written.
# post_write_hooks = black
# black.type = console_scripts
# black.entrypoint = black
# black.options = -l 79 REVISION_SCRIPT_FILENAME

# Logging configuration
[loggers]
keys = root,sqlalchemy,alembic

[handlers]
keys = console

[formatters]
keys = generic

[logger_root]
level = WARN
handlers = console
qualname =

[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine

[logger_alembic]
level = INFO
handlers =
qualname = alembic

[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic

[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S

Why This Configuration: The env.py file connects Alembic to your SQLAlchemy models, enabling autogenerate functionality. The alembic.ini file provides global configuration including database URLs and logging settings.

SHA-Based Revision IDs (Production Strategy)¶

# migrations/env.py - Enhanced with SHA-based revision generation
import hashlib
import uuid
from datetime import datetime

def generate_revision_id():
    """Generate SHA-based revision ID with timestamp and UUID"""
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    unique_id = str(uuid.uuid4()).replace('-', '')[:8]
    content = f"{timestamp}_{unique_id}"
    return f"{timestamp}_{hashlib.sha256(content.encode()).hexdigest()[:8]}"

# Use in migration creation
def run_migrations_online():
    # ... existing code ...

    # Set custom revision ID generation
    context.configure(
        connection=connection,
        target_metadata=target_metadata,
        compare_type=True,
        compare_server_default=True,
        # Custom revision ID generation
        revision_id_generator=lambda **kwargs: generate_revision_id(),
    )

Migration File Naming Convention¶

# File naming: YYYY-MM-DD_description.py
# Examples:
# 2024-01-15_create_users_table.py
# 2024-01-15_add_user_profile_fields.py
# 2024-01-15_migrate_user_data.py
# 2024-01-15_rename_posts_to_articles.py

# Revision ID format: YYYYMMDD_HHMMSS_sha8
# Examples:
# 20240115_103000_a1b2c3d4
# 20240115_110000_b2c3d4e5
# 20240115_120000_c3d4e5f6

Why SHA-Based IDs: Sequential revision IDs cause conflicts in distributed teams. SHA-based IDs with timestamps provide uniqueness, traceability, and prevent merge conflicts. The YYYY-MM-DD file naming convention ensures chronological ordering and human readability.

4) Creating & Running Migrations (The Workflow)¶

Autogenerate Migrations¶

# Create initial migration
alembic revision --autogenerate -m "create users and posts tables"

# Create migration for model changes
alembic revision --autogenerate -m "add user profile fields"

# Create migration with SHA-based revision ID
alembic revision --autogenerate -m "add indexes" --rev-id $(date +%Y%m%d_%H%M%S)_$(echo -n "$(date +%Y%m%d_%H%M%S)_add_indexes" | sha256sum | cut -c1-8)

Example Autogenerated Migration¶

# migrations/versions/2024-01-15_create_users_and_posts_tables.py
"""create users and posts tables

Revision ID: 20240115_103000_a1b2c3d4
Revises: 
Create Date: 2024-01-15 10:30:00.000000

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

# revision identifiers, used by Alembic.
revision = '20240115_103000_a1b2c3d4'
down_revision = None
branch_labels = None
depends_on = None

def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('users',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('username', sa.String(length=50), nullable=False),
    sa.Column('email', sa.String(length=100), nullable=False),
    sa.Column('full_name', sa.String(length=100), nullable=True),
    sa.Column('is_active', sa.Boolean(), nullable=True),
    sa.Column('created_at', sa.DateTime(), nullable=True),
    sa.Column('updated_at', sa.DateTime(), nullable=True),
    sa.PrimaryKeyConstraint('id'),
    sa.UniqueConstraint('email'),
    sa.UniqueConstraint('username')
    )
    op.create_table('posts',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('title', sa.String(length=200), nullable=False),
    sa.Column('content', sa.Text(), nullable=True),
    sa.Column('author_id', sa.Integer(), nullable=False),
    sa.Column('is_published', sa.Boolean(), nullable=True),
    sa.Column('created_at', sa.DateTime(), nullable=True),
    sa.Column('updated_at', sa.DateTime(), nullable=True),
    sa.ForeignKeyConstraint(['author_id'], ['users.id'], ),
    sa.PrimaryKeyConstraint('id')
    )
    # ### end Alembic commands ###

def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_table('posts')
    op.drop_table('users')
    # ### end Alembic commands ###

Apply Migrations¶

# Apply all pending migrations
alembic upgrade head

# Apply specific migration
alembic upgrade 20240115_103000_a1b2c3d4

# Apply migrations up to a specific revision
alembic upgrade 20240115_120000_b2c3d4e5

# Show current revision
alembic current

# Show migration history
alembic history

# Show pending migrations
alembic show head

Downgrade Migrations¶

# Downgrade one step
alembic downgrade -1

# Downgrade to specific revision
alembic downgrade 20240115_103000_a1b2c3d4

# Downgrade to base (empty database)
alembic downgrade base

# Show what would be downgraded
alembic downgrade -1 --sql

Why This Workflow: Autogenerate creates migrations based on model changes, while manual review ensures correctness. The upgrade/downgrade cycle enables safe schema evolution with rollback capabilities.

5) Writing Manual Migrations (The Art)¶

Custom Migration Example¶

# migrations/versions/2024-01-15_add_user_profile_fields.py
"""add user profile fields

Revision ID: 20240115_110000_b2c3d4e5
Revises: 20240115_103000_a1b2c3d4
Create Date: 2024-01-15 11:00:00.000000

"""
from alembic import op
import sqlalchemy as sa

# revision identifiers, used by Alembic.
revision = '20240115_110000_b2c3d4e5'
down_revision = '20240115_103000_a1b2c3d4'
branch_labels = None
depends_on = None

def upgrade() -> None:
    # Add new columns
    op.add_column('users', sa.Column('bio', sa.Text(), nullable=True))
    op.add_column('users', sa.Column('avatar_url', sa.String(length=255), nullable=True))
    op.add_column('users', sa.Column('date_of_birth', sa.Date(), nullable=True))

    # Add index for performance
    op.create_index('ix_users_email', 'users', ['email'])

    # Add check constraint
    op.create_check_constraint(
        'ck_users_bio_length',
        'users',
        'LENGTH(bio) <= 1000'
    )

def downgrade() -> None:
    # Remove constraints and indexes first
    op.drop_constraint('ck_users_bio_length', 'users', type_='check')
    op.drop_index('ix_users_email', 'users')

    # Remove columns
    op.drop_column('users', 'date_of_birth')
    op.drop_column('users', 'avatar_url')
    op.drop_column('users', 'bio')

Data Migration Example¶

# migrations/versions/2024-01-15_migrate_user_data.py
"""migrate user data

Revision ID: 20240115_120000_c3d4e5f6
Revises: 20240115_110000_b2c3d4e5
Create Date: 2024-01-15 12:00:00.000000

"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy import text

# revision identifiers, used by Alembic.
revision = '20240115_120000_c3d4e5f6'
down_revision = '20240115_110000_b2c3d4e5'
branch_labels = None
depends_on = None

def upgrade() -> None:
    # Get database connection
    connection = op.get_bind()

    # Update existing users with default bio
    connection.execute(
        text("UPDATE users SET bio = 'No bio available' WHERE bio IS NULL")
    )

    # Set default avatar for users without one
    connection.execute(
        text("UPDATE users SET avatar_url = '/default-avatar.png' WHERE avatar_url IS NULL")
    )

    # Create user statistics table
    op.create_table('user_stats',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('user_id', sa.Integer(), nullable=False),
        sa.Column('login_count', sa.Integer(), default=0),
        sa.Column('last_login', sa.DateTime(), nullable=True),
        sa.Column('created_at', sa.DateTime(), nullable=True),
        sa.PrimaryKeyConstraint('id'),
        sa.ForeignKeyConstraint(['user_id'], ['users.id'], ),
    )

    # Populate user stats for existing users
    connection.execute(
        text("""
            INSERT INTO user_stats (user_id, login_count, last_login, created_at)
            SELECT id, 0, created_at, created_at FROM users
        """)
    )

def downgrade() -> None:
    # Remove user stats table
    op.drop_table('user_stats')

    # Reset bio and avatar_url to NULL
    connection = op.get_bind()
    connection.execute(text("UPDATE users SET bio = NULL, avatar_url = NULL"))

Complex Schema Changes¶

# migrations/versions/2024-01-15_rename_and_restructure.py
"""rename and restructure tables

Revision ID: 20240115_130000_d4e5f6g7
Revises: 20240115_120000_c3d4e5f6
Create Date: 2024-01-15 13:00:00.000000

"""
from alembic import op
import sqlalchemy as sa

# revision identifiers, used by Alembic.
revision = '20240115_130000_d4e5f6g7'
down_revision = '20240115_120000_c3d4e5f6'
branch_labels = None
depends_on = None

def upgrade() -> None:
    # Rename table
    op.rename_table('posts', 'articles')

    # Rename column
    op.alter_column('articles', 'title', new_column_name='headline')

    # Change column type
    op.alter_column('articles', 'content',
        existing_type=sa.Text(),
        type_=sa.String(length=5000),
        existing_nullable=True
    )

    # Add new column with default
    op.add_column('articles', sa.Column('word_count', sa.Integer(), default=0))

    # Update word_count for existing articles
    connection = op.get_bind()
    connection.execute(
        sa.text("UPDATE articles SET word_count = LENGTH(content)")
    )

def downgrade() -> None:
    # Reverse the changes
    op.alter_column('articles', 'headline', new_column_name='title')
    op.alter_column('articles', 'content',
        existing_type=sa.String(length=5000),
        type_=sa.Text(),
        existing_nullable=True
    )
    op.drop_column('articles', 'word_count')
    op.rename_table('articles', 'posts')

Why Manual Migrations: Some schema changes require custom SQL or data transformations that autogenerate cannot handle. Manual migrations provide full control over the migration process.

6) Migration Flow Diagram¶

flowchart TD
    A[SQLAlchemy Models] -->|alembic revision --autogenerate| B[Migration Script]
    B -->|Review & Edit| C[Final Migration]
    C -->|alembic upgrade head| D[Database Schema Updated]
    D --> E[Application Uses New Schema]

    F[Manual Changes] -->|alembic revision -m| G[Custom Migration]
    G -->|Edit with SQL| H[Final Migration]
    H -->|alembic upgrade head| D

    I[Production Deploy] -->|alembic upgrade head| J[Schema Updated in Production]
    J --> K[Application Restart]
    K --> L[New Features Available]

    M[Rollback Needed] -->|alembic downgrade -1| N[Previous Schema]
    N --> O[Application Restart]
    O --> P[Previous Version Running]

    style A fill:#e1f5fe
    style B fill:#f3e5f5
    style C fill:#e8f5e8
    style D fill:#fff3e0
    style E fill:#fce4ec

Why This Flow: The migration flow shows the complete lifecycle from model changes to production deployment. Understanding this flow is essential for managing database evolution safely.

7) Managing Environments (The Strategy)¶

Environment-Specific Configuration¶

# config.py
import os
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

class Config:
    # Database URLs for different environments
    DATABASE_URLS = {
        'development': 'postgresql://user:pass@localhost/myapp_dev',
        'testing': 'postgresql://user:pass@localhost/myapp_test',
        'staging': 'postgresql://user:pass@staging-db/myapp_staging',
        'production': 'postgresql://user:pass@prod-db/myapp_prod'
    }

    @classmethod
    def get_database_url(cls, environment='development'):
        return cls.DATABASE_URLS.get(environment)

    @classmethod
    def create_engine(cls, environment='development'):
        url = cls.get_database_url(environment)
        return create_engine(url, echo=True)

# Environment-specific alembic.ini files
# alembic_dev.ini
[alembic]
script_location = migrations
sqlalchemy.url = postgresql://user:pass@localhost/myapp_dev

# alembic_staging.ini
[alembic]
script_location = migrations
sqlalchemy.url = postgresql://user:pass@staging-db/myapp_staging

# alembic_prod.ini
[alembic]
script_location = migrations
sqlalchemy.url = postgresql://user:pass@prod-db/myapp_prod

Environment-Specific Commands¶

# Development
alembic -c alembic_dev.ini upgrade head

# Staging
alembic -c alembic_staging.ini upgrade head

# Production
alembic -c alembic_prod.ini upgrade head

# Using environment variables
export DATABASE_URL="postgresql://user:pass@localhost/myapp_dev"
alembic upgrade head

CI/CD Integration¶

# .github/workflows/deploy.yml
name: Deploy with Migrations

on:
  push:
    branches: [main]

jobs:
  deploy:
    runs-on: ubuntu-latest
    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 alembic

      - name: Run migrations
        env:
          DATABASE_URL: ${{ secrets.DATABASE_URL }}
        run: |
          alembic upgrade head

      - name: Deploy application
        run: |
          # Your deployment commands here

Why Environment Management: Different environments require different database configurations. Proper environment management ensures migrations run correctly across development, staging, and production.

8) Best Practices (The Wisdom)¶

Migration Best Practices¶

# 1. Always review autogenerated migrations
# 2. Keep migrations in version control
# 3. Never regenerate existing migrations
# 4. Write idempotent data migrations
# 5. Test downgrade paths

# Example: Idempotent data migration
def upgrade() -> None:
    connection = op.get_bind()

    # Check if column already exists
    inspector = sa.inspect(connection)
    if 'new_column' not in inspector.get_columns('users'):
        op.add_column('users', sa.Column('new_column', sa.String(100)))

    # Update only if needed
    result = connection.execute(sa.text("SELECT COUNT(*) FROM users WHERE new_column IS NULL"))
    if result.scalar() > 0:
        connection.execute(sa.text("UPDATE users SET new_column = 'default' WHERE new_column IS NULL"))

def downgrade() -> None:
    # Safe rollback
    connection = op.get_bind()
    inspector = sa.inspect(connection)
    if 'new_column' in inspector.get_columns('users'):
        op.drop_column('users', 'new_column')

Branching and Merging¶

# Handle diverging migration histories
# 1. Create merge migration with SHA-based IDs
alembic merge -m "merge user and post changes" 20240115_103000_a1b2c3d4 20240115_110000_b2c3d4e5

# 2. Resolve conflicts in the merge migration
# 3. Apply the merge migration
alembic upgrade head

# Example merge migration
def upgrade() -> None:
    # Apply changes from both branches
    op.add_column('users', sa.Column('profile_data', sa.JSON()))
    op.add_column('posts', sa.Column('tags', sa.ARRAY(sa.String())))

    # Handle any conflicts
    connection = op.get_bind()
    # Custom logic to resolve conflicts

Performance Considerations¶

# Large table migrations
def upgrade() -> None:
    # Add column with default value (PostgreSQL)
    op.add_column('large_table', sa.Column('new_field', sa.String(100), default=''))

    # Update in batches for large tables
    connection = op.get_bind()
    batch_size = 10000

    while True:
        result = connection.execute(
            sa.text("SELECT id FROM large_table WHERE new_field = '' LIMIT :batch_size"),
            {'batch_size': batch_size}
        )
        rows = result.fetchall()
        if not rows:
            break

        ids = [row[0] for row in rows]
        connection.execute(
            sa.text("UPDATE large_table SET new_field = 'default' WHERE id = ANY(:ids)"),
            {'ids': ids}
        )

Why These Practices: Following best practices ensures reliable migrations, prevents data loss, and enables safe rollbacks. These practices are essential for production database management.

9) Common Pitfalls (The Traps)¶

Autogenerate Limitations¶

# Autogenerate may miss:
# 1. Default values from server-side functions
# 2. Custom constraints
# 3. Indexes not defined in models
# 4. Triggers and stored procedures

# Example: Manual fix for missing defaults
def upgrade() -> None:
    # Autogenerate missed the default value
    op.add_column('users', sa.Column('created_at', sa.DateTime(), nullable=False))

    # Manually add the default
    connection = op.get_bind()
    connection.execute(
        sa.text("ALTER TABLE users ALTER COLUMN created_at SET DEFAULT CURRENT_TIMESTAMP")
    )

Rename Detection Issues¶

# Alembic may show renames as drop/add
# Fix manually in the migration

def upgrade() -> None:
    # Instead of drop/add, use rename
    op.rename_table('old_table', 'new_table')
    op.alter_column('new_table', 'old_column', new_column_name='new_column')

SQLite Limitations¶

# SQLite has limited ALTER TABLE support
# Use batch operations for complex changes

def upgrade() -> None:
    # SQLite doesn't support dropping columns
    # Use batch operations
    with op.batch_alter_table('users') as batch_op:
        batch_op.drop_column('old_column')
        batch_op.add_column('new_column', sa.String(100))

Data Loss Prevention¶

# Always backup before major migrations
def upgrade() -> None:
    # Create backup table
    op.execute("CREATE TABLE users_backup AS SELECT * FROM users")

    # Perform risky operations
    op.drop_column('users', 'risky_column')

    # Verify data integrity
    connection = op.get_bind()
    result = connection.execute(sa.text("SELECT COUNT(*) FROM users_backup"))
    backup_count = result.scalar()

    result = connection.execute(sa.text("SELECT COUNT(*) FROM users"))
    current_count = result.scalar()

    if backup_count != current_count:
        raise Exception("Data loss detected!")

Why These Pitfalls Matter: Understanding common pitfalls prevents data loss and migration failures. These examples show how to handle Alembic's limitations and edge cases.

10) Advanced Features (The Power)¶

Custom Migration Context¶

# migrations/env.py - Advanced configuration
def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        # Custom context configuration
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            compare_type=True,
            compare_server_default=True,
            # Custom options
            include_object=include_object,
            include_schemas=True,
            version_table_schema='public',
        )

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

def include_object(object, name, type_, reflected, compare_to):
    """Custom filter for migration objects"""
    if type_ == "table" and name.startswith("temp_"):
        return False
    return True

Migration Hooks¶

# migrations/env.py - Add hooks
def run_migrations_online():
    # ... existing code ...

    with context.begin_transaction():
        # Pre-migration hook
        context.execute("SET lock_timeout = '30s'")
        context.run_migrations()
        # Post-migration hook
        context.execute("ANALYZE")

Custom Migration Templates¶

# migrations/script.py.mako - Custom template
"""${message}

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

"""
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:
    ${upgrades if upgrades else "pass"}

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

Why Advanced Features: Advanced Alembic features provide fine-grained control over migrations. These features enable complex schema changes and custom migration logic.

11) TL;DR Quickstart (The Essentials)¶

Essential Commands¶

# 1. Install and initialize
pip install alembic sqlalchemy psycopg2-binary
alembic init migrations

# 2. Configure env.py with your models and SHA-based revision IDs
# Edit migrations/env.py to import your Base.metadata and add revision_id_generator

# 3. Create initial migration with SHA-based ID
alembic revision --autogenerate -m "initial migration" --rev-id $(date +%Y%m%d_%H%M%S)_$(echo -n "$(date +%Y%m%d_%H%M%S)_initial_migration" | sha256sum | cut -c1-8)

# 4. Apply migrations
alembic upgrade head

# 5. Check status
alembic current
alembic history

# 6. Rollback if needed
alembic downgrade -1

Essential Configuration¶

# migrations/env.py - Minimal setup with SHA-based revision IDs
import hashlib
import uuid
from datetime import datetime
from alembic import context
from myapp.models import Base

target_metadata = Base.metadata

def generate_revision_id():
    """Generate SHA-based revision ID with timestamp and UUID"""
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    unique_id = str(uuid.uuid4()).replace('-', '')[:8]
    content = f"{timestamp}_{unique_id}"
    return f"{timestamp}_{hashlib.sha256(content.encode()).hexdigest()[:8]}"

def run_migrations_offline():
    url = context.config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url, 
        target_metadata=target_metadata,
        revision_id_generator=lambda **kwargs: generate_revision_id()
    )
    with context.begin_transaction():
        context.run_migrations()

def run_migrations_online():
    connectable = engine_from_config(
        context.config.get_section(context.config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    with connectable.connect() as connection:
        context.configure(
            connection=connection, 
            target_metadata=target_metadata,
            revision_id_generator=lambda **kwargs: generate_revision_id()
        )
        with context.begin_transaction():
            context.run_migrations()

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

Essential Workflow¶

# Daily workflow with SHA-based revision IDs
# 1. Make model changes
# 2. Generate migration with SHA-based ID
alembic revision --autogenerate -m "describe changes" --rev-id $(date +%Y%m%d_%H%M%S)_$(echo -n "$(date +%Y%m%d_%H%M%S)_describe_changes" | sha256sum | cut -c1-8)

# 3. Review and edit migration
# 4. Test locally
alembic upgrade head

# 5. Test rollback
alembic downgrade -1
alembic upgrade head

# 6. Commit to version control (migration files follow YYYY-MM-DD_description.py format)
git add migrations/versions/$(date +%Y-%m-%d)_*.py
git commit -m "Add migration for feature X"

# 7. Deploy to production
alembic upgrade head

Why This Quickstart: These commands cover 90% of daily Alembic usage. Master these before exploring advanced features.

12) The Machine's Summary¶

Alembic migrations are the foundation of reliable database schema management. When configured properly, they provide version control for your database, enable safe deployments, and prevent data loss. The key is understanding the migration lifecycle, mastering the workflow, and following best practices.

The Dark Truth: Without migrations, your database schema is a ticking time bomb. Alembic migrations are your safety net. Use them wisely.

The Machine's Mantra: "In migrations we trust, in version control we build, and in the database we find the path to reliable deployments."

Why This Matters: Database schema evolution is inevitable. Alembic migrations provide the tools to manage this evolution safely, enabling confident deployments and reliable rollbacks.

This tutorial provides the complete machinery for managing database evolution with Alembic. The patterns scale from development to production, from simple schema changes to complex data migrations.