Alembic

Alembic Cheatsheet

Installation & Setup

# Install
pip install alembic

# Initialize
alembic init alembic
alembic init -t async alembic  # For async support

Configuration

alembic.ini

sqlalchemy.url = postgresql://user:pass@localhost/dbname

# env.py - Use environment variable
config.set_main_option('sqlalchemy.url', os.environ.get('DATABASE_URL'))

---

Migration Commands

# Create migrations
alembic revision --autogenerate -m "description"  # Auto-generate
alembic revision -m "description"                 # Empty migration

# Apply migrations
alembic upgrade head      # Latest version
alembic upgrade +1        # Next version
alembic upgrade <rev_id>  # Specific version

# Rollback migrations
alembic downgrade -1      # Previous version
alembic downgrade <rev_id> # Specific version
alembic downgrade base    # Rollback all

# Information
alembic current           # Show current revision
alembic history           # Show history
alembic history --verbose # Detailed history
alembic heads             # Show pending migrations
alembic branches          # Show branches

# Stamping
alembic stamp head        # Mark as current without running
alembic stamp <rev_id>    # Mark specific revision

# Generate SQL (offline mode)
alembic upgrade head --sql > migration.sql
alembic downgrade -1 --sql > rollback.sql

---

Table Operations

# Create table
op.create_table(
    'users',
    sa.Column('id', sa.Integer, primary_key=True),
    sa.Column('name', sa.String(50), nullable=False),
    sa.Column('email', sa.String(100), unique=True),
    sa.Column('created_at', sa.DateTime, server_default=sa.func.now())
)

# Drop table
op.drop_table('users')

# Rename table
op.rename_table('old_name', 'new_name')

---

Column Operations

# Add column
op.add_column('users', sa.Column('age', sa.Integer, nullable=True))

# Drop column
op.drop_column('users', 'age')

# Alter column
op.alter_column('users', 'name',
                type_=sa.String(100),
                nullable=False,
                new_column_name='full_name',
                server_default='Unknown')

# Rename column
op.alter_column('users', 'old_name', new_column_name='new_name')

---

Index Operations

# Create index
op.create_index('idx_email', 'users', ['email'])
op.create_index('idx_name_age', 'users', ['name', 'age'])  # Composite

# Drop index
op.drop_index('idx_email', 'users')

# Create unique constraint
op.create_unique_constraint('uq_email', 'users', ['email'])

# Drop constraint
op.drop_constraint('uq_email', 'users', type_='unique')

---

Foreign Key Operations

# Add foreign key
op.create_foreign_key(
    'fk_user_id',           # Constraint name
    'orders',               # Source table
    'users',                # Target table
    ['user_id'],            # Source columns
    ['id'],                 # Target columns
    ondelete='CASCADE'      # Optional
)

# Drop foreign key
op.drop_constraint('fk_user_id', 'orders', type_='foreignkey')

---

Enum Operations (PostgreSQL)

Create Enum

from sqlalchemy.dialects import postgresql

# Method 1: Raw SQL
op.execute("CREATE TYPE status_enum AS ENUM ('pending', 'active', 'inactive')")

# Method 2: SQLAlchemy
status_enum = postgresql.ENUM('pending', 'active', 'inactive', name='status_enum')
status_enum.create(op.get_bind(), checkfirst=True)

# Add column with enum
op.add_column('users', sa.Column('status', status_enum, nullable=False))

Add Enum Value

# Simple add
op.execute("ALTER TYPE status_enum ADD VALUE 'suspended'")

# With position (PostgreSQL 12+)
op.execute("ALTER TYPE status_enum ADD VALUE 'archived' AFTER 'inactive'")

# Safe add (if not exists)
op.execute("""
    DO $$ BEGIN
        IF NOT EXISTS (
            SELECT 1 FROM pg_enum WHERE enumlabel = 'suspended'
            AND enumtypid = (SELECT oid FROM pg_type WHERE typname = 'status_enum')
        ) THEN ALTER TYPE status_enum ADD VALUE 'suspended';
        END IF;
    END $$;
""")

Rename Enum

# Rename type
op.execute("ALTER TYPE status_enum RENAME TO user_status")

# Rename value (PostgreSQL 10+)
op.execute("ALTER TYPE status_enum RENAME VALUE 'inactive' TO 'disabled'")

Remove Enum Value (Recreate Pattern)

# Create new → Copy data → Replace old
op.execute("CREATE TYPE status_new AS ENUM ('pending', 'active')")
op.add_column('users', sa.Column('status_temp',
              postgresql.ENUM('pending', 'active', name='status_new')))
op.execute("UPDATE users SET status_temp = status::text::status_new")
op.drop_column('users', 'status')
op.execute('DROP TYPE status_enum')
op.execute('ALTER TYPE status_new RENAME TO status_enum')
op.alter_column('users', 'status_temp', new_column_name='status')

Drop Enum

op.drop_column('users', 'status')  # Drop column first
op.execute('DROP TYPE status_enum')

Cross-Database Enum

import enum

class Status(enum.Enum):
    PENDING = "pending"
    ACTIVE = "active"

# Native enum in PostgreSQL, VARCHAR in others
op.add_column('users',
    sa.Column('status', sa.Enum(Status, native_enum=True),
              nullable=False, server_default='pending'))

---

Enum Operations (MySQL)

# Create column with enum
op.execute("""
    ALTER TABLE users
    ADD COLUMN status ENUM('pending', 'active', 'inactive')
    NOT NULL DEFAULT 'pending'
""")

# Modify enum values (add/remove)
op.execute("""
    ALTER TABLE users
    MODIFY COLUMN status ENUM('pending', 'active', 'suspended')
    NOT NULL DEFAULT 'pending'
""")

# Drop enum column
op.drop_column('users', 'status')

---

Data Migrations

from sqlalchemy import table, column

# Define table structure
users_table = table('users',
    column('id', sa.Integer),
    column('name', sa.String)
)

# Bulk insert
op.bulk_insert(users_table, [
    {'id': 1, 'name': 'Alice'},
    {'id': 2, 'name': 'Bob'}
])

# Update
op.execute(
    users_table.update()
    .where(users_table.c.id == 1)
    .values(name='Alice Updated')
)

# Delete
op.execute(
    users_table.delete()
    .where(users_table.c.id == 1)
)

# Raw SQL
op.execute("UPDATE users SET active = true WHERE created_at > '2024-01-01'")
op.execute(sa.text("UPDATE users SET status = :status"), {'status': 'active'})

---

Advanced Features

Multiple Databases

alembic -n database1 upgrade head
alembic -n database2 revision -m "description"

Branching & Merging

alembic revision -m "branch" --head=<parent>  # Create branch
alembic merge -m "merge" <rev1> <rev2>        # Merge branches
alembic branches                               # Show branches

Transaction Control

def upgrade():
    # Operations in a transaction
    with op.batch_alter_table('users') as batch_op:
        batch_op.add_column(sa.Column('age', sa.Integer))
        batch_op.create_index('idx_age', ['age'])

Conditional Migrations

from alembic import context

def upgrade():
    conn = op.get_bind()

    # Check if column exists
    inspector = sa.inspect(conn)
    columns = [col['name'] for col in inspector.get_columns('users')]

    if 'age' not in columns:
        op.add_column('users', sa.Column('age', sa.Integer))

---

Migration File Structure

"""Add user status

Revision ID: abc123
Revises: def456
Create Date: 2024-01-01 12:00:00
"""
from alembic import op
import sqlalchemy as sa

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

def upgrade():
    """Apply changes"""
    pass

def downgrade():
    """Revert changes"""
    pass

---

Quick Reference Table

Operation	Command
Create Enum	CREATE TYPE name AS ENUM (...)
Add Enum Value	ALTER TYPE name ADD VALUE 'val'
Remove Enum Value	❌ Recreate type required
Rename Enum	ALTER TYPE old RENAME TO new
Drop Enum	DROP TYPE name
Create Table	op.create_table('name', ...)
Add Column	op.add_column('table', sa.Column(...))
Create Index	op.create_index('name', 'table', ['col'])
Add FK	op.create_foreign_key(...)
Execute SQL	op.execute('SQL')

---

Best Practices

✅ DO

• Review auto-generated migrations
• Test on dev/staging first
• Keep migrations small and focused
• Include both upgrade and downgrade
• Use meaningful descriptions
• Backup database before production migrations
• Use checkfirst=True for enum types
• Clean up enum types in downgrade

❌ DON’T

• Edit applied migrations (create new ones)
• Skip downgrade implementation
• Use migrations for application logic
• Hardcode sensitive data
• Remove enum values directly (PostgreSQL)
• Forget to drop enum types in cleanup

---

Troubleshooting

# Debug mode (show SQL)
alembic upgrade head --sql

# Check current state
alembic current

# Reset to version without running
alembic stamp <revision_id>

# Show specific revision details
alembic show <revision_id>

# Check enum values
SELECT enumlabel FROM pg_enum e
JOIN pg_type t ON e.enumtypid = t.oid
WHERE t.typname = 'status_enum';