alembic

Alembic

Overview

Alembic is the migration tool for SQLAlchemy. It records schema changes as versioned Python scripts, each pointing at its parent revision, and stores the current revision in an alembic_version table. It can autogenerate scripts by comparing your models with the live database, run data migrations, handle branches, and emit plain SQL for review. Checked against Alembic 1.20 (September 2026), which requires Python 3.10+.

Instructions

Step 1: Setup

pip install alembic
alembic list_templates                 # generic, async, multidb, pyproject, pyproject_async
alembic init migrations                # sync driver (psycopg, pymysql, sqlite)
alembic init -t async migrations       # async driver (asyncpg, aiosqlite, aiomysql)

Set the database URL in alembic.ini (sqlalchemy.url) or override it in env.py from an environment variable such as DATABASE_URL; do not commit passwords. Then point env.py at your models:

# migrations/env.py (edit the generated file, do not replace it)
from app.models import Base          # your DeclarativeBase subclass
target_metadata = Base.metadata

The async template already contains the async engine wiring and the offline mode branch (needed for --sql). Async drivers need pip install "sqlalchemy[asyncio]" asyncpg (or aiosqlite); without greenlet, env.py fails with an ImportError. Hand-written env.py files that always call run_migrations_online() break --sql, so start from the template.

Step 2: Create migrations

alembic revision --autogenerate -m "add projects table"   # diff models vs database
alembic revision -m "backfill user roles"                 # empty script for data work
alembic check                                             # fails if models differ from the migrations

Generated files are named <hash>_<slug>.py and look like this (autogenerate output, adjusted by hand):

def upgrade() -> None:
    op.create_table(
        "projects",
        sa.Column("id", sa.Integer(), primary_key=True),
        sa.Column("name", sa.String(100), nullable=False),
        sa.Column("owner_id", sa.Integer(), sa.ForeignKey("users.id"), nullable=False),
    )
    op.create_index(op.f("ix_projects_owner_id"), "projects", ["owner_id"])

def downgrade() -> None:
    op.drop_index(op.f("ix_projects_owner_id"), table_name="projects")
    op.drop_table("projects")