Alembic - database migration

Introduction

The evolution of modern web applications inevitably ties to the evolution of database schemas. Adding new tables, changing column types, creating indexes, and modifying relationships between entities — all require a systematic approach to schema management. Manual schema changes are prone to errors, data loss, and inconsistencies across development environments.

Alembic is a powerful database migration tool built specifically for the SQLAlchemy ecosystem. As the official solution, it provides safe and controlled schema changes, allowing developers to track migration history, apply updates, and roll back problematic modifications when needed.

What Is Alembic

Alembic — a migration library for Python projects that use SQLAlchemy. It offers tools to create, apply, and revert database schema changes in a controlled order.

Key Benefits of Alembic

Schema versioning — each change is recorded as a separate migration with a unique identifier, enabling full visibility of data model evolution.

Automatic migration generation — ability to generate migration files from changes in SQLAlchemy ORM models, dramatically speeding up development.

Operation safety — all changes run inside transactions, minimizing the risk of data corruption.

Multi‑DBMS support — works with PostgreSQL, MySQL, SQLite, Oracle, and any other database supported by SQLAlchemy.

Installation and Configuration

Package Installation

pip install alembic

For asynchronous applications, install the extra dependencies as well:

pip install alembic[asyncio]

Initializing a Migration Project

After installation, bootstrap the migration directory structure:

alembic init alembic

The command creates the following layout:

alembic/
├── versions/           # Directory for migration scripts
├── env.py              # Environment configuration
├── script.py.mako      # Template used to generate migrations
└── alembic.ini         # Main configuration file

Database Connection Settings

Specify the database URL in alembic.ini:

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

Alternatively, configure the connection programmatically in env.py:

from sqlalchemy import create_engine
from myproject.config import DATABASE_URL

config.set_main_option("sqlalchemy.url", DATABASE_URL)

Integration with SQLAlchemy

Linking Models

To enable autogeneration, expose the model metadata in alembic/env.py:

from myproject.models import Base

# Provide metadata for autogeneration
target_metadata = Base.metadata

Sample Base Model

from sqlalchemy import Column, Integer, String, DateTime, Boolean
from sqlalchemy.ext.declarative import declarative_base
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)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)

Core Alembic Commands

Creating Migrations

Applying and Reverting Migrations

Information Commands

Creating and Applying Migrations

Manual Migration Creation

alembic revision -m "create users table"

Then edit the generated script:

from alembic import op
import sqlalchemy as sa

def upgrade():
    op.create_table(
        'users',
        sa.Column('id', sa.Integer, primary_key=True),
        sa.Column('username', sa.String(50), nullable=False),
        sa.Column('email', sa.String(100), nullable=False),
        sa.Column('is_active', sa.Boolean, default=True),
        sa.Column('created_at', sa.DateTime, default=sa.func.now())
    )
    op.create_index('ix_users_username', 'users', ['username'])
    op.create_index('ix_users_email', 'users', ['email'])

def downgrade():
    op.drop_index('ix_users_email', 'users')
    op.drop_index('ix_users_username', 'users')
    op.drop_table('users')

Automatic Migration Generation

alembic revision --autogenerate -m "add email column to users"

Alembic compares the current Base.metadata with the live database and generates the required operations.

Operations in Migrations

Core Operations Table

Complex Migration Example

def upgrade():
    # Create a new table
    op.create_table(
        'posts',
        sa.Column('id', sa.Integer, primary_key=True),
        sa.Column('title', sa.String(200), nullable=False),
        sa.Column('content', sa.Text),
        sa.Column('user_id', sa.Integer, nullable=False),
        sa.Column('created_at', sa.DateTime, default=sa.func.now())
    )
    
    # Add a foreign key
    op.create_foreign_key(
        'fk_posts_user_id',
        'posts', 'users',
        ['user_id'], ['id']
    )
    
    # Create indexes
    op.create_index('ix_posts_title', 'posts', ['title'])
    op.create_index('ix_posts_user_id', 'posts', ['user_id'])
    
    # Add a new column to an existing table
    op.add_column('users', sa.Column('last_login', sa.DateTime))

def downgrade():
    op.drop_column('users', 'last_login')
    op.drop_index('ix_posts_user_id', 'posts')
    op.drop_index('ix_posts_title', 'posts')
    op.drop_constraint('fk_posts_user_id', 'posts', type_='foreignkey')
    op.drop_table('posts')

Working with Asynchronous Projects

Async Configuration

SQLAlchemy 2.0 brings stable async support. To use an async engine, modify env.py as follows:

from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.pool import NullPool
import asyncio

def run_migrations_online():
    connectable = create_async_engine(
        config.get_main_option("sqlalchemy.url"),
        poolclass=NullPool,
        future=True
    )
    
    async def do_run_migrations(connection):
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            literal_binds=True,
            compare_type=True
        )
        
        with context.begin_transaction():
            context.run_migrations()
    
    async def run_async_migrations():
        async with connectable.connect() as connection:
            await connection.run_sync(do_run_migrations)
    
    asyncio.run(run_async_migrations())

Async Model Example

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String

Base = declarative_base()

class AsyncUser(Base):
    __tablename__ = 'async_users'
    
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True)
    email = Column(String(100), unique=True)

Version Management and History

Branching Migrations

Alembic supports branching, which is useful for multiple development teams:

# Create a new branch
alembic revision -m "feature branch" --branch-label feature

# Merge branches
alembic merge -m "merge feature branch" --branch-label master feature

Viewing Migration History

# Detailed history
alembic history --verbose

# Reverse order
alembic history -r -1:

# History for a specific branch
alembic history -r feature:

Testing Migrations

Creating a Test Database

import pytest
from alembic.config import Config
from alembic import command
from sqlalchemy import create_engine

@pytest.fixture
def test_db():
    # Create an in‑memory SQLite database
    engine = create_engine("sqlite:///:memory:")
    
    # Apply migrations
    alembic_cfg = Config("alembic.ini")
    alembic_cfg.set_main_option("sqlalchemy.url", "sqlite:///:memory:")
    
    with engine.begin() as connection:
        alembic_cfg.attributes['connection'] = connection
        command.upgrade(alembic_cfg, "head")
    
    yield engine
    engine.dispose()

Verifying Migration Correctness

def test_migration_upgrade_downgrade(test_db):
    """Test applying and rolling back migrations"""
    alembic_cfg = Config("alembic.ini")
    
    # Apply migrations
    command.upgrade(alembic_cfg, "head")
    
    # Verify current version exists
    current_rev = command.current(alembic_cfg)
    assert current_rev is not None
    
    # Roll back migrations
    command.downgrade(alembic_cfg, "base")
    
    # Verify rollback
    current_rev = command.current(alembic_cfg)
    assert current_rev is None

CI/CD Integration

GitHub Actions

name: Database Migrations
on:
  push:
    branches: [ main ]
  pull_request:
    branches: [ main ]

jobs:
  test-migrations:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:13
        env:
          POSTGRES_PASSWORD: postgres
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
    steps:
    - uses: actions/checkout@v2
    
    - name: Set up Python
      uses: actions/setup-python@v2
      with:
        python-version: 3.9
    
    - name: Install dependencies
      run: |
        pip install -r requirements.txt
    
    - name: Run migrations
      run: |
        alembic upgrade head
      env:
        DATABASE_URL: postgresql://postgres:postgres@localhost/test_db
    
    - name: Test migration rollback
      run: |
        alembic downgrade -1
        alembic upgrade head

GitLab CI

stages:
  - test
  - deploy

test-migrations:
  stage: test
  image: python:3.9
  services:
    - postgres:13
  variables:
    POSTGRES_DB: test_db
    POSTGRES_USER: postgres
    POSTGRES_PASSWORD: postgres
  script:
    - pip install -r requirements.txt
    - alembic upgrade head
    - alembic downgrade base
  only:
    - merge_requests
    - main

deploy-migrations:
  stage: deploy
  script:
    - alembic upgrade head
  only:
    - main
  when: manual

Working with Multiple Databases

Multi‑DB Configuration

# env.py
from sqlalchemy import create_engine
import os

def get_engine_for_environment():
    environment = os.getenv('ENVIRONMENT', 'development')
    
    if environment == 'production':
        return create_engine(os.getenv('PRODUCTION_DATABASE_URL'))
    elif environment == 'staging':
        return create_engine(os.getenv('STAGING_DATABASE_URL'))
    else:
        return create_engine(os.getenv('DEVELOPMENT_DATABASE_URL'))

def run_migrations_online():
    connectable = get_engine_for_environment()
    
    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata
        )
        
        with context.begin_transaction():
            context.run_migrations()

Configuration Management

# config.py
import os
from dataclasses import dataclass

@dataclass
class DatabaseConfig:
    url: str
    echo: bool = False
    pool_size: int = 5

class Config:
    def __init__(self):
        self.environment = os.getenv('ENVIRONMENT', 'development')
        
    def get_database_config(self) -> DatabaseConfig:
        configs = {
            'development': DatabaseConfig(
                url=os.getenv('DEV_DATABASE_URL'),
                echo=True
            ),
            'testing': DatabaseConfig(
                url=os.getenv('TEST_DATABASE_URL'),
                echo=False
            ),
            'production': DatabaseConfig(
                url=os.getenv('PROD_DATABASE_URL'),
                echo=False,
                pool_size=20
            )
        }
        return configs.get(self.environment, configs['development'])

Best Practices and Recommendations

Migration Naming

Use descriptive names — instead of “update users,” use “add_email_verification_to_users.”

Follow a consistent convention — apply the same naming style across the team.

Include context — add information about which part of the system is affected.

Migration Structure

One migration per change — avoid bundling unrelated changes together.

Review autogenerated code — always inspect generated scripts before applying them.

Add comments — explain complex alterations and the rationale behind them.

Operation Safety

def upgrade():
    # Check for table existence before creation
    bind = op.get_bind()
    inspector = inspect(bind)
    
    if 'users' not in inspector.get_table_names():
        op.create_table(
            'users',
            sa.Column('id', sa.Integer, primary_key=True),
            sa.Column('username', sa.String(50), nullable=False)
        )
    
    # Safely add a column if it does not exist
    if 'email' not in [col['name'] for col in inspector.get_columns('users')]:
        op.add_column('users', sa.Column('email', sa.String(100)))

Comparison with Other Tools

Common Issues and Solutions

Fixing “Target database is not up to date”

# Check current version
alembic current

# Apply missing migrations
alembic upgrade head

# Or stamp the database as up‑to‑date
alembic stamp head

When a Migration Doesn’t Apply

# Inspect the generated SQL
alembic upgrade head --sql

# Apply a specific revision
alembic upgrade <revision_id>

# Skip a problematic migration
alembic stamp <revision_id>

Rolling Back Multiple Migrations

# Revert to a specific revision
alembic downgrade <revision_id>

# Step back several revisions
alembic downgrade -3

# Full rollback to base
alembic downgrade base

Resolving Migration Conflicts

# Create a merge migration
alembic merge -m "merge conflicting branches" <rev1> <rev2>

# Apply the merged migration
alembic upgrade head

Monitoring and Logging

Logging Configuration

import logging
from alembic.config import Config
from alembic import command

# Basic Alembic logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('alembic')

def run_migrations_with_logging():
    alembic_cfg = Config("alembic.ini")
    
    # Attach a custom logger
    def log_migration(rev, context):
        logger.info(f"Applying migration: {rev}")
    
    alembic_cfg.attributes['logger'] = logger
    command.upgrade(alembic_cfg, "head")

Migration Status Monitoring

from alembic.config import Config
from alembic.script import ScriptDirectory
from alembic.runtime.migration import MigrationContext
from sqlalchemy import create_engine

def get_migration_status():
    alembic_cfg = Config("alembic.ini")
    script = ScriptDirectory.from_config(alembic_cfg)
    
    engine = create_engine(alembic_cfg.get_main_option("sqlalchemy.url"))
    with engine.connect() as connection:
        context = MigrationContext.configure(connection)
        current_rev = context.get_current_revision()
        head_rev = script.get_current_head()
        
        return {
            'current_revision': current_rev,
            'head_revision': head_rev,
            'is_up_to_date': current_rev == head_rev
        }

Conclusion

Alembic is an indispensable tool for managing database migrations in Python projects that use SQLAlchemy. Its flexibility, reliability, and rich feature set make it the ideal choice for applications of any size — from small services to large enterprise solutions.

Proper use of Alembic ensures controlled schema evolution, reduces deployment risk, and streamlines collaboration across teams. Automatic migration generation accelerates development, while rollback capabilities safeguard production environments.

Integration with modern development workflows, support for asynchronous operations, and compatibility with a wide range of DBMSs position Alembic as the de‑facto standard for migration management in the Python ecosystem.