alembic

Alembic

Overview

Alembic is the migration tool for SQLAlchemy. It tracks database schema changes as versioned Python scripts — like Git for your database. Supports autogeneration from model changes, branching, and data migrations.

Instructions

Step 1: Setup

pip install alembic
alembic init alembic

# alembic/env.py — Configure with async SQLAlchemy
from alembic import context
from sqlalchemy.ext.asyncio import create_async_engine
from models import Base
import asyncio

config = context.config
target_metadata = Base.metadata

def run_migrations_online():
    connectable = create_async_engine(config.get_main_option("sqlalchemy.url"))

    async def do_run():
        async with connectable.connect() as connection:
            await connection.run_sync(do_migrations)

    def do_migrations(connection):
        context.configure(connection=connection, target_metadata=target_metadata)
        with context.begin_transaction():
            context.run_migrations()

    asyncio.run(do_run())

run_migrations_online()

Step 2: Create Migrations

# Auto-generate from model changes
alembic revision --autogenerate -m "add projects table"

# Create empty migration (for data migrations)
alembic revision -m "backfill user roles"

# alembic/versions/001_add_projects.py — Generated migration
def upgrade():
    op.create_table('projects',
        sa.Column('id', sa.String(36), primary_key=True),
        sa.Column('name', sa.String(100), nullable=False),
        sa.Column('owner_id', sa.String(36), sa.ForeignKey('users.id')),
        sa.Column('created_at', sa.DateTime, server_default=sa.func.now()),
    )
    op.create_index('ix_projects_owner_id', 'projects', ['owner_id'])

def downgrade():
    op.drop_index('ix_projects_owner_id')
    op.drop_table('projects')

Step 3: Data Migrations

# alembic/versions/002_backfill_roles.py — Data migration
from alembic import op
import sqlalchemy as sa

def upgrade():
    # Add column
    op.add_column('users', sa.Column('role', sa.String(20), server_default='member'))

    # Backfill existing rows
    conn = op.get_bind()
    conn.execute(sa.text("UPDATE users SET role = 'admin' WHERE email LIKE '%@mycompany.com'"))

def downgrade():
    op.drop_column('users', 'role')