Alembic: Database Migration Management for SQLAlchemy

When your application grows, your database schema evolves. You add columns, rename tables, or change constraints. Without a migration tool, you'd manually run SQL scripts, risking inconsistencies across environments. Alembic, created by the author of SQLAlchemy, solves this by providing a version-controlled, repeatable way to manage schema changes.

Alembic generates migration scripts that describe how to transform your database from one version to the next. It integrates tightly with SQLAlchemy models, auto-detecting changes and producing the necessary SQL. This tutorial will guide you from setup to production debugging, including a real-world incident that taught a team the hard way about branching migrations.

By the end, you'll be able to confidently manage database migrations in any Python project, whether it's a Flask API, Django app (with SQLAlchemy), or a FastAPI service.

Setting Up Alembic

First, install Alembic and SQLAlchemy. Then initialize Alembic in your project directory. This creates a migrations folder with an env.py file that configures the migration environment. You need to point Alembic to your SQLAlchemy database URL and metadata. Here's a typical setup:

``bash pip install alembic sqlalchemy alembic init migrations ``

Edit alembic.ini to set your database URL:

``ini sqlalchemy.url = postgresql://user:pass@localhost/mydb ``

In migrations/env.py, import your SQLAlchemy models and set target_metadata:

``python from myapp.models import Base target_metadata = Base.metadata ``

Now you can create your first migration: alembic revision --autogenerate -m "initial". This scans your models and generates a script with upgrade() and downgrade() functions.

# migrations/env.py
from alembic import context
from sqlalchemy import engine_from_config
from myapp.models import Base  # your SQLAlchemy models

target_metadata = Base.metadata

def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix='sqlalchemy.')
    with connectable.connect() as connection:
        context.configure(connection=connection,
                          target_metadata=target_metadata)
        with context.begin_transaction():
            context.run_migrations()

Understanding Migration Scripts

Each migration script is a Python file with upgrade() and downgrade() functions. These functions use Alembic's Operations API to modify the schema. Common operations include create_table, add_column, alter_column, and create_index. Here's an example of a migration that adds a profile_pic column:

```python """add profile_pic column

Revision ID: abc123 Revises: def456 Create Date: 2025-03-15 10:00:00 """ from alembic import op import sqlalchemy as sa

revision = 'abc123' down_revision = 'def456'

def upgrade(): op.add_column('user', sa.Column('profile_pic', sa.String(200), nullable=True))

def downgrade(): op.drop_column('user', 'profile_pic') ```

The down_revision links scripts in a chain. Alembic uses this to know the order. You can also use branch_labels for branching migrations (e.g., for feature branches).

# migrations/versions/abc123_add_profile_pic.py
"""add profile_pic column

Revision ID: abc123
Revises: def456
Create Date: 2025-03-15 10:00:00
"""
from alembic import op
import sqlalchemy as sa

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

def upgrade():
    op.add_column('user', sa.Column('profile_pic', sa.String(200), nullable=True))

def downgrade():
    op.drop_column('user', 'profile_pic')

Autogenerating Migrations

Alembic's --autogenerate compares your SQLAlchemy model metadata with the current database state and generates migration scripts. This is the recommended way to create migrations. However, autogenerate cannot detect everything:

• Can detect: new tables, columns, column type changes, indexes, unique constraints, foreign keys.
• Cannot detect: table/column renames, data migrations, changes to server defaults, or constraints like CHECK.

For these, you must manually edit the generated script. Always review autogenerated scripts before applying them. Example:

``bash alembic revision --autogenerate -m "add email column" ``

This will produce a script with the detected changes. If you rename a column, autogenerate will see it as a drop + add, which would lose data. So you must manually change it to a rename operation.

# After running alembic revision --autogenerate -m "add email"
# The generated script might look like:

def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column('user', sa.Column('email', sa.String(), nullable=True))
    # ### end Alembic commands ###

def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_column('user', 'email')
    # ### end Alembic commands ###

Applying and Rolling Back Migrations

To apply all pending migrations, run alembic upgrade head. To roll back the last migration, use alembic downgrade -1. You can also specify a revision ID: alembic upgrade abc123. Alembic tracks the current revision in the alembic_version table.

It's crucial to test downgrades. For example, if you add a NOT NULL column, the downgrade should handle existing data. You can use batch operations for complex changes:

```python def upgrade(): with op.batch_alter_table('user') as batch_op: batch_op.add_column(sa.Column('age', sa.Integer(), nullable=False, server_default='0'))

def downgrade(): with op.batch_alter_table('user') as batch_op: batch_op.drop_column('age') ```

Batch mode is especially useful for SQLite, which doesn't support ALTER TABLE.

"""add age column with default

Revision ID: xyz789
Revises: abc123
"""
from alembic import op
import sqlalchemy as sa

revision = 'xyz789'
down_revision = 'abc123'

def upgrade():
    with op.batch_alter_table('user') as batch_op:
        batch_op.add_column(sa.Column('age', sa.Integer(), nullable=False, server_default='0'))

def downgrade():
    with op.batch_alter_table('user') as batch_op:
        batch_op.drop_column('age')

Branching and Merging Migrations

When multiple developers work on the same project, they may create migration scripts based on the same parent revision. This creates a branch. Alembic supports branching via branch_labels. To resolve branches, you create a merge migration that combines them.

Example: Two developers both create migrations from revision 'abc'. Developer A creates 'def', developer B creates 'ghi'. The migration history becomes a fork. To merge:

``bash alembic merge -m "merge def and ghi" def ghi ``

This creates a new migration with down_revision pointing to both 'def' and 'ghi'. The merge script's upgrade() should be empty (or contain any needed conflict resolution). The downgrade() is usually a no-op.

Branching is powerful but can lead to complex histories. Use feature branches with caution.

"""merge def and ghi

Revision ID: merge123
Revises: def, ghi
Create Date: 2025-03-15 12:00:00
"""
from alembic import op
import sqlalchemy as sa

revision = 'merge123'
down_revision = ('def', 'ghi')  # tuple of parent revisions

def upgrade():
    pass

def downgrade():
    pass

Data Migrations and Custom Operations

Sometimes you need to transform existing data during a migration, e.g., populate a new column based on another column. Alembic allows you to execute raw SQL or use SQLAlchemy Core within migration scripts. Example:

```python def upgrade(): op.add_column('user', sa.Column('full_name', sa.String())) op.execute("UPDATE user SET full_name = first_name || ' ' || last_name")

def downgrade(): op.drop_column('user', 'full_name') ```

For complex data migrations, use SQLAlchemy's table() to reflect the table and perform bulk updates. Be careful with large datasets; consider batching.

You can also create custom operations by subclassing MigrateOperation. This is advanced but useful for reusable patterns.

"""populate full_name from first and last name

Revision ID: data123
Revises: xyz789
"""
from alembic import op
import sqlalchemy as sa

revision = 'data123'
down_revision = 'xyz789'

def upgrade():
    op.add_column('user', sa.Column('full_name', sa.String()))
    # Use SQLAlchemy Core for bulk update
    from sqlalchemy.sql import table, column
    user = table('user', column('full_name'), column('first_name'), column('last_name'))
    op.execute(
        user.update().values(
            full_name=user.c.first_name + ' ' + user.c.last_name
        )
    )

def downgrade():
    op.drop_column('user', 'full_name')

Production Best Practices

1. Version control: Always commit migration scripts. Never generate them on production.
2. CI checks: Run alembic check to ensure models match the latest migration. Also run alembic upgrade head in CI against a test database.
3. Idempotency: Write migrations that can be run multiple times safely (use if not column_exists checks).
4. Backup: Before any migration in production, take a database backup.
5. Rollback plan: Test downgrade scripts. Have a manual rollback procedure.
6. Environment-specific: Use different migration branches for different environments if needed, but prefer a linear history.
7. Locking: Long-running migrations can lock tables. Use --sql to generate SQL and run manually with care.
8. Monitoring: Log migration output and monitor for errors. Use tools like alembic current in health checks.

Example CI script:

``yaml # .github/workflows/migrations.yml jobs: test-migrations: runs-on: ubuntu-latest services: postgres: image: postgres:15 env: POSTGRES_PASSWORD: test options: >- --health-cmd pg_isready --health-interval 10s steps: - uses: actions/checkout@v3 - uses: actions/setup-python@v4 - run: pip install -r requirements.txt - run: alembic upgrade head env: DATABASE_URL: postgresql://postgres:test@localhost:5432/postgres ``

# In your CI script, you can also run a Python check:
import subprocess
result = subprocess.run(['alembic', 'check'], capture_output=True, text=True)
if result.returncode != 0:
    print("Migrations are not up-to-date!")
    print(result.stdout)
    exit(1)
else:
    print("Database schema is up-to-date.")