Alembic Textbook Ch. 3 — Writing Migrations Manually

Chapter 3: writing migrations by hand. The op API, common operations, data migrations, batch alters.

Empty migration

alembic revision -m "add admin role"

Creates a stub. Fill in upgrade() / downgrade().

op API basics

from alembic import op
import sqlalchemy as sa

def upgrade():
    op.create_table(
        "items",
        sa.Column("id", sa.Integer, primary_key=True),
        sa.Column("name", sa.String(120), nullable=False),
        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
    )
    op.create_index("ix_items_name", "items", ["name"])

def downgrade():
    op.drop_index("ix_items_name", "items")
    op.drop_table("items")

op.* provides all DDL operations.

Common operations

Add column

op.add_column("users", sa.Column("status", sa.String(20), nullable=False, server_default="active"))

Drop column

op.drop_column("users", "old_field")

Alter column

op.alter_column("users", "name",
    type_=sa.String(200),
    nullable=False,
    new_column_name="full_name",
)

Rename table

op.rename_table("old_name", "new_name")

Indexes

op.create_index("ix_users_email", "users", ["email"], unique=True)
op.drop_index("ix_users_email", "users")

Constraints

op.create_unique_constraint("uq_users_email", "users", ["email"])
op.create_check_constraint("ck_users_age_positive", "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")

Raw SQL

op.execute("UPDATE users SET status = 'active' WHERE status IS NULL")
op.execute("CREATE INDEX CONCURRENTLY ix_users_email ON users (email)")

For DB-specific or complex SQL.

Data migrations

def upgrade():
    op.add_column("users", sa.Column("status", sa.String(20), nullable=True))
    
    # Bind a connection for inserts/updates
    bind = op.get_bind()
    bind.execute(sa.text("UPDATE users SET status = 'active'"))
    
    op.alter_column("users", "status", nullable=False)

op.get_bind() returns the current connection. Use SQLAlchemy text or table objects.

For large tables: don’t load into Python. Use SQL:

op.execute("UPDATE users SET status = 'active'")

For really large tables: break into chunks (see Chapter 5).

Using SQLAlchemy in migrations

For data migration with structured access:

def upgrade():
    op.add_column("users", sa.Column("active", sa.Boolean, nullable=True))
    
    # Define a lightweight table for the migration
    users_table = sa.table(
        "users",
        sa.column("id", sa.Integer),
        sa.column("status", sa.String),
        sa.column("active", sa.Boolean),
    )
    
    bind = op.get_bind()
    bind.execute(
        users_table.update().values(active=(users_table.c.status == "active"))
    )
    
    op.alter_column("users", "active", nullable=False)

Don’t import your real model classes — they may have evolved. Define a frozen table inside the migration.

Conditional logic

def upgrade():
    bind = op.get_bind()
    inspector = sa.inspect(bind)
    
    if "users" in inspector.get_table_names():
        # ...
    
    cols = [c["name"] for c in inspector.get_columns("users")]
    if "old_field" in cols:
        op.drop_column("users", "old_field")

Inspect existing schema. Useful for migrations that have been partially applied.

batch_alter_table (SQLite)

SQLite doesn’t support most ALTER:

with op.batch_alter_table("users") as batch:
    batch.add_column(sa.Column("status", sa.String(20)))
    batch.alter_column("name", new_column_name="full_name")

Alembic emulates: creates new table, copies data, drops old, renames.

For Postgres / MySQL: not needed (native ALTER).

Postgres-specific

Concurrent indexes

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

def downgrade():
    op.execute("DROP INDEX CONCURRENTLY ix_users_email")

CONCURRENTLY can’t run in a transaction:

# alembic.ini or env.py
[alembic]
transaction_per_migration = false

Or per-migration setting:

def upgrade():
    op.execute("COMMIT")
    op.execute("CREATE INDEX CONCURRENTLY ...")
    op.execute("BEGIN")  # restart for the rest

Awkward. Better: transaction_per_migration = false for the whole project.

Extensions

def upgrade():
    op.execute("CREATE EXTENSION IF NOT EXISTS pg_trgm")

ENUM

Adding a value:

op.execute("ALTER TYPE user_role ADD VALUE 'editor'")

Note: can’t be in a transaction in older PG. PG 12+: yes, but not undoable.

For schema-flexible: use VARCHAR + CHECK instead of native ENUM.

Triggers

def upgrade():
    op.execute("""
        CREATE OR REPLACE FUNCTION update_modified_column()
        RETURNS TRIGGER AS $$
        BEGIN
            NEW.updated_at = now();
            RETURN NEW;
        END;
        $$ LANGUAGE plpgsql;
    """)
    op.execute("""
        CREATE TRIGGER set_updated_at
        BEFORE UPDATE ON users
        FOR EACH ROW
        EXECUTE FUNCTION update_modified_column();
    """)

def downgrade():
    op.execute("DROP TRIGGER IF EXISTS set_updated_at ON users")
    op.execute("DROP FUNCTION IF EXISTS update_modified_column")

Idempotent migrations

For migrations that may run on different schema states:

def upgrade():
    op.execute("""
        DO $$
        BEGIN
            IF NOT EXISTS (SELECT 1 FROM pg_indexes WHERE indexname = 'ix_users_email') THEN
                CREATE INDEX ix_users_email ON users (email);
            END IF;
        END
        $$;
    """)

PG-specific. For DB-agnostic: inspector check.

Generally: Alembic’s revision tracking ensures one run per migration. Idempotency is for edge cases.

Long-running migrations

For migrations that take hours: split into multiple revisions or run as a separate process while app is running.

# 1. Add nullable column
def upgrade():
    op.add_column("orders", sa.Column("status", sa.String(20), nullable=True))

# Then deploy code that handles NULL status as "pending"
# Then run a one-off backfill outside Alembic in batches
# Then:

# 2. Make NOT NULL
def upgrade():
    op.alter_column("orders", "status", nullable=False)

See Database Migrations 2026 .

Common mistakes

1. Importing app models in migration

Migration runs against future schema; imports may fail. Use frozen tables.

2. Big UPDATE

UPDATE users SET status = ... on 100M rows: hours of lock. Batch outside Alembic.

3. Non-CONCURRENTLY index on hot table

Locks. Use CONCURRENTLY.

4. Forgetting downgrade

Stub with pass is OK in prod. Make sure dev can roll back.

5. Mixing DDL and DML in transactions awkwardly

Some DDL (CONCURRENTLY, certain ENUM ops) can’t run in transactions. Configure.

What’s next

Chapter 4: branching and merging.

Empty migration#

op API basics#

Common operations#

Add column#

Drop column#

Alter column#

Rename table#

Indexes#

Constraints#

Raw SQL#

Data migrations#

Using SQLAlchemy in migrations#

Conditional logic#

batch_alter_table (SQLite)#

Postgres-specific#

Concurrent indexes#

Extensions#

ENUM#

Triggers#

Idempotent migrations#

Long-running migrations#

Common mistakes#

1. Importing app models in migration#

2. Big UPDATE#

3. Non-CONCURRENTLY index on hot table#

4. Forgetting downgrade#

5. Mixing DDL and DML in transactions awkwardly#

What’s next#

Read this next#