SQLAlchemy Cheatsheet 09 — Migrations Integration (Alembic)

alembic init -t async migrations

[alembic]
script_location = migrations
sqlalchemy.url =                # leave empty; set from env.py

import asyncio
from sqlalchemy.ext.asyncio import async_engine_from_config
from alembic import context

# Import all model modules so Base.metadata is populated
import src.app.models  # noqa

from src.app.db import Base
from src.app.settings import settings

config = context.config
config.set_main_option("sqlalchemy.url", settings.database_url)

target_metadata = Base.metadata

def do_run_migrations(connection):
    context.configure(
        connection=connection,
        target_metadata=target_metadata,
        compare_type=True,                   # detect type changes
        compare_server_default=True,         # detect default changes
        include_schemas=False,               # toggle for multi-schema
    )
    with context.begin_transaction():
        context.run_migrations()

async def run_migrations_online():
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
    )
    async with connectable.connect() as conn:
        await conn.run_sync(do_run_migrations)
    await connectable.dispose()

if context.is_offline_mode():
    raise NotImplementedError
else:
    asyncio.run(run_migrations_online())

NAMING_CONVENTION = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s",
}

class Base(DeclarativeBase):
    metadata = MetaData(naming_convention=NAMING_CONVENTION)

alembic revision --autogenerate -m "msg"
alembic upgrade head
alembic upgrade +1
alembic downgrade -1
alembic current
alembic history
alembic heads
alembic show <rev>
alembic stamp head        # mark current without running

[ $(alembic heads | wc -l) -gt 1 ] && exit 1

def test_no_drift(engine):
    from alembic.autogenerate import compare_metadata
    from alembic.migration import MigrationContext
    
    with engine.sync_engine.connect() as conn:
        ctx = MigrationContext.configure(conn)
        diff = compare_metadata(ctx, Base.metadata)
        assert diff == [], f"Drift: {diff}"

from alembic import op
import sqlalchemy as sa

op.create_table("users", ...)
op.drop_table("users")
op.add_column("users", sa.Column("status", sa.String(20), nullable=False, server_default="active"))
op.drop_column("users", "status")
op.alter_column("users", "name", type_=sa.String(200), nullable=False, new_column_name="full_name")
op.rename_table("old", "new")
op.create_index("ix_users_email", "users", ["email"], unique=True)
op.drop_index("ix_users_email", "users")
op.create_unique_constraint("uq_users_email", "users", ["email"])
op.create_check_constraint("ck_users_age", "users", "age > 0")
op.create_foreign_key("fk_posts_user", "posts", "users", ["user_id"], ["id"], ondelete="CASCADE")
op.drop_constraint("fk_posts_user", "posts", type_="foreignkey")
op.execute("SQL ...")

def upgrade():
    op.execute("CREATE INDEX CONCURRENTLY ix_users_email ON users (email)")

[alembic]
transaction_per_migration = false

def upgrade():
    op.add_column("users", sa.Column("status", sa.String(20), nullable=True))
    
    # Lightweight table inside the migration (don't import ORM models)
    users = sa.table("users",
        sa.column("id", sa.Integer),
        sa.column("status", sa.String),
    )
    
    bind = op.get_bind()
    bind.execute(users.update().values(status="active"))
    
    op.alter_column("users", "status", nullable=False)

apiVersion: batch/v1
kind: Job
metadata: { name: migrate-{{ .Values.image.tag }} }
spec:
  template:
    spec:
      containers:
        - name: migrate
          image: myapi:{{ .Values.image.tag }}
          command: ["alembic", "upgrade", "head"]
      restartPolicy: OnFailure

Phase 1 (expand): add new column; backfill; code writes both.
Phase 2 (contract): code uses new only; drop old.