Alembic in production: safe patterns

Capsule overview

You already have the individual techniques: expand-contract (capsule 03), CREATE INDEX CONCURRENTLY (capsule 04), lock_timeout and statement_timeout (capsule 05). Now you're going to learn how they integrate into a complete production workflow. The difference between "I know Alembic" and "I know Alembic in 24/7 multi-tenant production" lives in four operational patterns that don't get documented in tutorials: separating the code deploy from the migration deploy, migration idempotence so they survive retries, configuring env.py for multi-environment production, and the runbook you need at 3am when a migration hangs in production.

This capsule is the module's most operational one. You're not going to learn new SQL — you're going to learn how to orchestrate what you already know into a reliable system. You're going to write idempotent migrations that can be re-run without breaking. You're going to configure env.py with global timeouts and transaction_per_migration=False where it applies. You're going to have a ready RUNBOOK-MIGRATION.md with exact commands to diagnose and resolve a hung migration in production.

By the end you'll have the set of patterns that makes the difference between a team that runs migrations "when there's time" and one that runs them as a weekly routine with confidence.

Mental model: the schema deploy vs the code deploy

In development, schema and code get deployed together: you stop the server, run alembic upgrade head, bring the server back up. Atomic, simple, no version coexistence.

In zero-downtime production, they're two separate events orchestrated independently:

Time →

[Code v1.0 serving] ────────────────────────────────────────────────►

         ↓ Migration runs (expand: add nullable column)
[Schema v1.0 + new nullable column] ────────────────────────────────►

                  ↓ Deploy v1.1 (rolling)
[Code v1.0 ─── v1.1 mix] ───── [Code v1.1 serving] ─────────────────►

                                        ↓ Backfill runs
                                  [Data backfilled] ────────────────►

                                              ↓ Deploy v1.2 (rolling)
                                        [Code v1.2 serving] ────────►

                                                    ↓ Migration runs (contract: SET NOT NULL)
                                              [Schema v1.0 with final NOT NULL] ►

Three schema events (expand, backfill, contract). Three code events (v1.0, v1.1, v1.2). Each runs independently, at convenient times, with its own potential rollback.

This demands several operational patterns:

1. Migrations do not run automatically with the code deploy. Run them at controlled moments (ideally under low traffic).
2. Migrations are idempotent — if the pipeline retries, the second execution doesn't break.
3. Backfills are separate from DDL migrations — they can run as standalone scripts, not in the deploy's critical path.
4. When something hangs, there's a clear runbook so you don't improvise at 3am.

Pattern 1: separate the code deploy from the migration deploy

The common anti-pattern: the app's Dockerfile runs alembic upgrade head when the container starts.

# ❌ ANTI-PATTERN
FROM python:3.12
COPY . /app
WORKDIR /app
RUN pip install -r requirements.txt
CMD alembic upgrade head && uvicorn app.main:app

Problems:

• If the migration takes 5 minutes, the container takes 5 minutes to be ready. Kubernetes/Docker can mark it "unhealthy" and kill it mid-migration, leaving inconsistent state.
• If you deploy N pods, all N try to run the migration at the same time. Even though Alembic has internal locking (the alembic_version table), the coordination is fragile.
• Dangerous migrations run at any moment (whenever you deploy), not at convenient times.
• There's no separation between "the new code is deployed" and "the migration was applied".

The correct pattern: migrations are explicit operations, separate from the code deploy.

Option A: the CI/CD pipeline runs migrations as a separate step

# .github/workflows/deploy.yml
jobs:
  build:
    # ... Docker image build

  run-migrations:
    needs: build
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      - name: Install dependencies
        run: pip install -r requirements.txt
      - name: Run migrations
        env:
          DATABASE_URL: ${{ secrets.PROD_DATABASE_URL }}
        run: alembic upgrade head

  deploy:
    needs: run-migrations
    # ... code deploy (rolling)

Pros: a single execution, before the deploy. If it fails, the deploy doesn't advance.

Cons: the migration runs on every deploy. For deploys that don't include a new migration, it's unnecessary overhead (Alembic detects there's nothing to apply and finishes fast, but there's still latency).

Option B: a manual job in GitHub Actions / equivalent

# .github/workflows/migrate.yml
on:
  workflow_dispatch:
    inputs:
      target_revision:
        description: 'Target revision (default: head)'
        required: false
        default: 'head'

jobs:
  migrate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Run migration
        env:
          DATABASE_URL: ${{ secrets.PROD_DATABASE_URL }}
        run: alembic upgrade ${{ github.event.inputs.target_revision }}

Pros: total control over when to run. A dangerous migration gets scheduled for a low-traffic window. It lets you specify a target revision (not always head).

Cons: it requires team discipline (remembering to run the job before the deploy). More human coordination.

Option C: a dedicated migration container

In Kubernetes, a dedicated Job:

# k8s/migration-job.yaml
apiVersion: batch/v1
kind: Job
metadata:
  name: db-migration-{{ .Values.gitSha }}
spec:
  template:
    spec:
      restartPolicy: Never
      containers:
        - name: migrate
          image: myapp:{{ .Values.gitSha }}
          command: ["alembic", "upgrade", "head"]
          env:
            - name: DATABASE_URL
              valueFrom:
                secretKeyRef:
                  name: db-creds
                  key: url
      backoffLimit: 0  # No automatic retry

Pros: native Kubernetes infrastructure. Centralized logs. Visible status.

Cons: it requires K8s setup. More components to maintain.

Recommendation: Option A (a CI/CD step) for small/medium teams. Option B (a manual job) for teams with discipline and frequent dangerous migrations. Option C for advanced K8s shops.

Pattern 2: migration idempotence

Why it matters: in real production, migrations sometimes run more than once.

• The deploy pipeline retries after a transient failure.
• The manual job gets run twice by mistake.
• After a rollback, the next deploy re-applies the migration.

If your migration isn't idempotent, the second attempt fails with column already exists, relation already exists, or similar errors. Idempotence makes the second execution safe (no effect if it was already applied).

Alembic has high-level idempotence via the alembic_version table: it marks which migrations were applied and doesn't re-apply them. But if a migration fails midway, that mark may not get updated, leaving the migration "partially applied".

Idempotence patterns at the SQL level

For CREATE / ADD:

def upgrade():
    op.execute("SET lock_timeout = '5s'")

    # ❌ BAD — fails if the column already exists
    op.add_column("tasks", sa.Column("priority", sa.Integer(), nullable=True))

    # ✅ GOOD — idempotent
    op.execute(
        "ALTER TABLE tasks ADD COLUMN IF NOT EXISTS priority INTEGER NULL"
    )

For CREATE INDEX:

# ✅ GOOD
op.execute(
    "CREATE INDEX IF NOT EXISTS idx_tasks_priority ON tasks (priority)"
)

# For CONCURRENTLY:
with op.get_context().autocommit_block():
    op.execute(
        "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_tasks_priority "
        "ON tasks (priority)"
    )

For DROP:

# ✅ GOOD
op.execute("ALTER TABLE tasks DROP COLUMN IF EXISTS legacy_status")
op.execute("DROP INDEX IF EXISTS idx_tasks_legacy_status")

For RENAME: (there's no IF NOT EXISTS, you have to check manually)

def upgrade():
    op.execute("SET lock_timeout = '5s'")
    op.execute(
        """
        DO $$
        BEGIN
            IF EXISTS (
                SELECT 1 FROM information_schema.columns
                WHERE table_name = 'tasks'
                  AND column_name = 'old_name'
            ) THEN
                ALTER TABLE tasks RENAME COLUMN old_name TO new_name;
            END IF;
        END $$;
        """
    )

For CONSTRAINTS:

def upgrade():
    op.execute(
        """
        DO $$
        BEGIN
            IF NOT EXISTS (
                SELECT 1 FROM pg_constraint
                WHERE conname = 'tasks_priority_check'
            ) THEN
                ALTER TABLE tasks
                ADD CONSTRAINT tasks_priority_check CHECK (priority >= 0);
            END IF;
        END $$;
        """
    )

For UPDATE (a backfill):

# The backfill is already idempotent by design (WHERE priority IS NULL)
# The second execution updates nothing because there are no NULLs left.

The "fully idempotent" migration pattern

# alembic/versions/00X_add_priority_idempotent.py
"""add tasks.priority column (idempotent)

This migration can be re-run with no adverse effects.
"""
from alembic import op
import sqlalchemy as sa


revision = "00X_priority"
down_revision = "previous"


def upgrade():
    op.execute("SET lock_timeout = '5s'")
    op.execute("SET LOCAL statement_timeout = '30s'")

    # Idempotent: IF NOT EXISTS
    op.execute(
        "ALTER TABLE tasks ADD COLUMN IF NOT EXISTS priority INTEGER NULL"
    )


def downgrade():
    op.execute("SET lock_timeout = '5s'")
    op.execute("SET LOCAL statement_timeout = '30s'")
    op.execute("ALTER TABLE tasks DROP COLUMN IF EXISTS priority")

Pattern 3: configure env.py for production

The alembic/env.py file is the heart of the configuration. In production there are several important settings the default template doesn't have.

The recommended base configuration

# alembic/env.py
"""Alembic environment for production."""
import os
from logging.config import fileConfig

from sqlalchemy import engine_from_config, pool
from alembic import context

from app.db.models import Base  # Your SQLAlchemy metadata

config = context.config

# Set DATABASE_URL from env
if database_url := os.environ.get("DATABASE_URL"):
    config.set_main_option("sqlalchemy.url", database_url)
else:
    raise RuntimeError("DATABASE_URL is not set")

if config.config_file_name is not None:
    fileConfig(config.config_file_name)

target_metadata = Base.metadata


def run_migrations_offline() -> None:
    """Run migrations in 'offline' mode (generates SQL without connecting)."""
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online() -> None:
    """Run migrations in 'online' mode (connects to the DB)."""
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,  # Don't reuse connections for migrations
        # Important defaults for production:
        connect_args={
            "options": (
                # Connection-level timeouts (sane defaults).
                # Each migration can override them with SET LOCAL.
                "-c lock_timeout=5000 "          # 5s
                "-c statement_timeout=60000 "    # 60s
                "-c idle_in_transaction_session_timeout=10000"  # 10s
            ),
        },
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            # Keep True (the default) unless you have a specific need
            transaction_per_migration=True,
            # Compare types in autogenerate (detects type changes)
            compare_type=True,
            # Compare server_default
            compare_server_default=True,
        )

        with context.begin_transaction():
            context.run_migrations()


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

Key decisions in this configuration:

1. poolclass=pool.NullPool: migrations don't need to reuse connections. NullPool creates a connection, executes, closes. No pool overhead and no risk of hung connections.

2. lock_timeout=5000 and statement_timeout=60000 at the connection level: sane defaults. Each dangerous migration overrides them with SET LOCAL if it needs something more permissive.

3. idle_in_transaction_session_timeout=10000: if the migration opens a transaction and sits idle for more than 10s (a bug, a deadlock), the transaction gets cancelled. It prevents zombie transactions from holding locks.

4. compare_type=True and compare_server_default=True: autogenerate improvements to detect subtle changes. It reduces the risk of "autogenerate produced an empty migration but the model changed".

5. transaction_per_migration=True: keep the default. When you need operations outside a transaction (CONCURRENTLY), use a targeted autocommit_block() (capsule 04).

Multi-environment (dev, staging, prod)

# alembic/env.py — multi-environment extension
ENV = os.environ.get("APP_ENV", "dev")

if ENV == "prod":
    DATABASE_URL = os.environ["PROD_DATABASE_URL"]
    LOCK_TIMEOUT_MS = 5000
    STATEMENT_TIMEOUT_MS = 60000
elif ENV == "staging":
    DATABASE_URL = os.environ["STAGING_DATABASE_URL"]
    LOCK_TIMEOUT_MS = 5000
    STATEMENT_TIMEOUT_MS = 120000  # More permissive in staging
elif ENV == "dev":
    DATABASE_URL = os.environ.get(
        "DATABASE_URL",
        "postgresql://localhost/myapp_dev",
    )
    LOCK_TIMEOUT_MS = 30000  # Very permissive in dev
    STATEMENT_TIMEOUT_MS = 0  # No limit in dev
else:
    raise ValueError(f"Invalid APP_ENV: {ENV}")

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

Pattern 4: the "hung migration" runbook

At 3am somebody wakes you up: "the migration we deployed 2 hours ago is still 'running'. The app is degraded. What do I do?"

Without a runbook: you improvise. You try random commands. You make things worse.

With a runbook: you run predefined steps in order. You resolve it in 10-30 minutes.

Here's the complete runbook. Memorize it or keep it accessible.

Step 1: identify the lock holder and the waiter

Connect to the DB with a psql client. Run:

-- Which locks are active on the app's tables?
SELECT
    pg_class.relname AS table_name,
    pg_locks.mode AS lock_mode,
    pg_locks.granted,
    pg_stat_activity.pid,
    pg_stat_activity.usename AS user,
    pg_stat_activity.application_name AS app,
    pg_stat_activity.state,
    age(now(), pg_stat_activity.query_start) AS duration,
    pg_stat_activity.query
FROM pg_locks
JOIN pg_class ON pg_locks.relation = pg_class.oid
JOIN pg_stat_activity ON pg_locks.pid = pg_stat_activity.pid
WHERE pg_class.relnamespace = 'public'::regnamespace
ORDER BY pg_locks.granted DESC, pg_stat_activity.query_start;

Example output:

 table_name |     lock_mode      | granted | pid  |  user  |    app    |  state  | duration |              query
------------+--------------------+---------+------+--------+-----------+---------+----------+----------------------------------
 tasks      | ACCESS EXCLUSIVE   | t       | 9999 | postgres | psql      | idle in transaction | 02:14:30 | LOCK TABLE tasks IN ...
 tasks      | ACCESS EXCLUSIVE   | f       | 1234 | app_user | alembic   | active  | 01:50:12 | ALTER TABLE tasks ADD COLUMN ...
 tasks      | ACCESS SHARE       | f       | 5678 | app_user | uvicorn   | active  | 01:50:00 | SELECT * FROM tasks WHERE id=42
 tasks      | ROW EXCLUSIVE      | f       | 5679 | app_user | uvicorn   | active  | 01:50:00 | INSERT INTO tasks ...

Reading the output:

• PID 9999 (the postgres user, psql, "idle in transaction" for 2:14): the lock holder that shouldn't be there. Some DBA or dev forgot to COMMIT in an interactive psql session.
• PID 1234 (alembic): your migration, waiting for the lock for 1:50.
• PIDs 5678, 5679 (uvicorn): app requests stuck behind the migration's lock.

Step 2: decide what to do

Option A: kill the holder (resolves most cases)

-- Cancel first (gentle)
SELECT pg_cancel_backend(9999);

-- If it doesn't respond, terminate (hard)
SELECT pg_terminate_backend(9999);

When it applies: the holder is clearly a runaway process (an idle psql session, an analytical query that isn't going to stop in time, a zombie transaction).

When it does NOT apply: the holder is another legitimate migration (rare but possible if coordination is broken), or a critical maintenance job.

Option B: kill your own migration

SELECT pg_cancel_backend(1234);

When it applies: you decide it's better to abort the migration and retry it later in a quiet window instead of keeping the deploy hung.

After the cancel: Alembic raises an exception, the transaction gets rolled back, the alembic_version mark doesn't get updated. The migration stays "pending". You can retry it with alembic upgrade head when you're ready.

Option C: wait (rarely correct)

When it applies: you know the holder is going to finish soon (e.g. a VACUUM that's close to done, an analytical query that's 95% complete according to metrics).

When it does NOT apply: most cases. If the holder has been hanging production for more than 30 minutes, the damage is already done.

Step 3: after resolving the lock

Once the lock is released:

• If the migration was waiting: now it can take the lock and proceed. If your lock_timeout is 5s, it already failed — you need to retry.
• If you cancelled the migration: clean state, retry when you're ready.

Check Alembic's state:

alembic current

Expected output: the revision where it stopped (not the one you were trying to apply). If it ended up "partially applied" (rare), you can:

• Mark it manually with alembic stamp <revision> (dangerous, only if you're sure of the state).
• Apply fixups with raw SQL and then alembic stamp.

Step 4: postmortem

After resolving the incident, analyze:

• What left the lock holder there at the start of the incident? How do you keep it from happening again?
• Did the migration have lock_timeout? If not, add it. If it did and the deploy hung anyway, review the deploy wrapper.
• Did the runbook work? Update it with the learnings.

The RUNBOOK-MIGRATION.md template

Here's the complete runbook you can copy into your repo:

# Runbook: hung migration in production

## Symptoms

- The deploy pipeline reports "migration running" for more than [X] minutes.
- The API shows elevated latency or 500 errors on endpoints touching the affected table.
- DB metrics show lock waiters piling up.

## Diagnosis (10 min)

### Step 1: identify active locks

Connect to the DB with psql using a role with read permissions on the catalog:

\`\`\`bash
psql $DATABASE_URL_PROD
\`\`\`

Run:

\`\`\`sql
SELECT
    pg_class.relname AS table_name,
    pg_locks.mode AS lock_mode,
    pg_locks.granted,
    pg_stat_activity.pid,
    pg_stat_activity.usename,
    pg_stat_activity.application_name,
    pg_stat_activity.state,
    age(now(), pg_stat_activity.query_start) AS duration,
    pg_stat_activity.query
FROM pg_locks
JOIN pg_class ON pg_locks.relation = pg_class.oid
JOIN pg_stat_activity ON pg_locks.pid = pg_stat_activity.pid
WHERE pg_class.relnamespace = 'public'::regnamespace
ORDER BY pg_locks.granted DESC, pg_stat_activity.query_start;
\`\`\`

### Step 2: identify the holder and the waiters

- **Holder:** the row with `granted = true` on the affected table.
- **Waiters:** the rows with `granted = false`.

Note down: the holder's PID, what query it has, how long it's been active.

## Decision (5 min)

### Case A: the holder is an idle session (psql, a dev forgot to COMMIT)

Indicators: `state = 'idle in transaction'`, `application_name = 'psql'`, the query is `BEGIN` or `LOCK TABLE`.

Action: terminate the holder.

\`\`\`sql
SELECT pg_terminate_backend([PID_HOLDER]);
\`\`\`

### Case B: the holder is a long analytical query

Indicators: `state = 'active'`, the query starts with `SELECT`, `application_name` points at a BI tool, it's been running a long time.

Action: coordinate with the analytics team. If there's no urgency, wait. If urgency is high, terminate:

\`\`\`sql
SELECT pg_cancel_backend([PID_HOLDER]);  -- Gentle first
\`\`\`

### Case C: the holder is the app itself

Indicators: `application_name = 'uvicorn'` or similar, the query is an app transaction.

Action: investigate why the transaction is hung. Possible causes: an undetected deadlock, badly written retry logic, a leaked connection.

\`\`\`sql
SELECT pg_terminate_backend([PID_HOLDER]);
\`\`\`

### Case D: nobody is the holder but the migration is hung

Indicators: the migration's query (the alembic PID) has been `active` for a long time, there's no `granted = false` waiting on anything.

Action: the migration is processing, not waiting. It may be legitimate (CREATE INDEX CONCURRENTLY takes minutes). Check with:

\`\`\`sql
-- See CREATE INDEX progress (PG 12+)
SELECT
    pid,
    phase,
    blocks_done,
    blocks_total,
    ROUND(blocks_done::numeric / NULLIF(blocks_total, 0) * 100, 1) AS pct
FROM pg_stat_progress_create_index
WHERE pid = [PID_MIGRATION];
\`\`\`

If there's progress, wait. If not, consider cancelling.

## Post-resolution (15 min)

### Step 1: confirm Alembic's state

\`\`\`bash
alembic current
\`\`\`

If it stayed at a revision before the attempted one: retry when you're ready.

### Step 2: verify integrity

- Did any index end up INVALID? (capsule 04, the pg_index query)
- Did any constraint end up NOT VALID without a VALIDATE? (`SELECT * FROM pg_constraint WHERE NOT convalidated;`)
- Is the application responding normally?

### Step 3: postmortem

Document in the team doc:
- The incident timeline
- The cause of the lock holder
- The action taken
- The learning (was the runbook enough? what's missing?)

Backfills as standalone scripts

An optional but important improvement for production: separating backfills from Alembic migrations into standalone scripts.

Why: backfills can take hours. Having them as a migration means alembic upgrade head hangs for those hours. As a standalone script, you run them separately (in the background, in a Kubernetes Job, etc.) without blocking the deploy.

Structure:

scripts/
├── backfills/
│   ├── 001_backfill_tasks_priority.py
│   └── 002_backfill_users_email_verified.py
└── ...

# scripts/backfills/001_backfill_tasks_priority.py
"""Backfill tasks.priority = 0 for rows with NULL.

Runs independently of the deploy pipeline.

Usage:
    python scripts/backfills/001_backfill_tasks_priority.py

State:
    The script is idempotent (WHERE priority IS NULL).
    If it gets interrupted, simply re-run it.
"""
import argparse
import os
import time

import psycopg2


DEFAULT_BATCH_SIZE = 10_000
DEFAULT_SLEEP_SEC = 0.1


def backfill(
    database_url: str,
    batch_size: int = DEFAULT_BATCH_SIZE,
    sleep_sec: float = DEFAULT_SLEEP_SEC,
    dry_run: bool = False,
) -> None:
    conn = psycopg2.connect(database_url)
    conn.autocommit = False

    try:
        with conn.cursor() as cur:
            # Session configuration
            cur.execute("SET lock_timeout = '5s'")
            cur.execute("SET statement_timeout = '0'")  # No total timeout

            # Get the ID range
            cur.execute(
                "SELECT COALESCE(MAX(id), 0) FROM tasks WHERE priority IS NULL"
            )
            max_id = cur.fetchone()[0] or 0

            if max_id == 0:
                print("Nothing to backfill.")
                return

            print(f"Backfill: max_id={max_id}, batch_size={batch_size}")

            batch_min = 0
            batches = 0
            rows_total = 0

            while batch_min <= max_id:
                batch_max = batch_min + batch_size - 1

                if dry_run:
                    cur.execute(
                        """
                        SELECT COUNT(*) FROM tasks
                        WHERE id BETWEEN %s AND %s AND priority IS NULL
                        """,
                        (batch_min, batch_max),
                    )
                    rows_in_batch = cur.fetchone()[0]
                else:
                    cur.execute(
                        """
                        UPDATE tasks SET priority = 0
                        WHERE id BETWEEN %s AND %s AND priority IS NULL
                        """,
                        (batch_min, batch_max),
                    )
                    rows_in_batch = cur.rowcount
                    conn.commit()  # Commit per batch (don't accumulate)

                rows_total += rows_in_batch
                batches += 1

                if batches % 10 == 0:
                    pct = min(100, (batch_min / max_id) * 100)
                    print(
                        f"Batch {batches}: progress {pct:.1f}% "
                        f"(total rows: {rows_total})"
                    )

                batch_min = batch_max + 1
                time.sleep(sleep_sec)

            # Final verification
            cur.execute("SELECT COUNT(*) FROM tasks WHERE priority IS NULL")
            null_remaining = cur.fetchone()[0]
            print(
                f"Backfill complete. Batches: {batches}, "
                f"rows updated: {rows_total}, "
                f"rows still NULL: {null_remaining}"
            )

            if null_remaining > 0 and not dry_run:
                print(
                    "WARNING: there are still rows with priority IS NULL. "
                    "Possible cause: the old app is still inserting NULLs. "
                    "Check the app deploy and re-run."
                )
    finally:
        conn.close()


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--batch-size", type=int, default=DEFAULT_BATCH_SIZE)
    parser.add_argument("--sleep", type=float, default=DEFAULT_SLEEP_SEC)
    parser.add_argument("--dry-run", action="store_true")
    args = parser.parse_args()

    database_url = os.environ.get("DATABASE_URL")
    if not database_url:
        raise SystemExit("DATABASE_URL is not set")

    backfill(
        database_url=database_url,
        batch_size=args.batch_size,
        sleep_sec=args.sleep,
        dry_run=args.dry_run,
    )

Usage in production:

# A dry run first to estimate the time
python scripts/backfills/001_backfill_tasks_priority.py --dry-run

# The real execution
python scripts/backfills/001_backfill_tasks_priority.py

# With more conservative batches if the DB is under load
python scripts/backfills/001_backfill_tasks_priority.py --batch-size 5000 --sleep 0.5

Advantages vs a backfill as an Alembic migration:

• It doesn't block the deploy pipeline.
• It can run at a convenient time, without coordinating with the deploy.
• Detailed logging to stdout, easy to monitor.
• Parameterizable (batch size, sleep) without a re-deploy.
• Idempotent and retryable.

Why does this matter in real work?

1. It's the knowledge that separates the senior from the lead. Knowing how to write migrations is senior. Knowing how to operate them in 24/7 multi-tenant production is lead. The patterns in this capsule are the ones a lead establishes as the team standard.

2. The hung-migration runbook is one of the seniority tests in interviews. "What do you do if the migration has been running for 2 hours in production?" is a standard question for senior+. Having the runbook articulated and being able to execute it under pressure sets you apart.

3. Backfills as standalone scripts is what separates teams that migrate without fear from teams paralyzed by fear of backfills. If your backfill blocks the deploy, you're going to postpone large migrations "until there's time." If it's separate, you do them as routine.

4. The env.py configuration this capsule teaches solves the most common operational problems you're going to hit in production: timeouts, idle transactions, type comparison in autogenerate. It's invisible work that pays dividends for years.

Traps and common mistakes

Mistake 1 (critical operational): running migrations at container startup

Symptom: the new pods take minutes to be ready. Kubernetes marks them unhealthy and kills them. Inconsistent state. A hung deploy.

Why it happens: the naive intuition of "the app needs the updated schema to start." The attempt to "automate it" at startup ends up causing the very problems it meant to avoid.

How to tell: review your Dockerfile. Is there an alembic upgrade in the CMD or ENTRYPOINT? If so, it's a potential problem.

How to fix it: move migrations to a separate pipeline step (Pattern 1 of this capsule). The container startup should be simple: bring up the server.

Mistake 2 (conceptual): thinking IF NOT EXISTS covers every form of idempotence

Symptom: the dev adds IF NOT EXISTS to a CREATE INDEX and calls it done. But the migration also does an ALTER COLUMN with no check, and the second attempt fails.

Why it happens: "idempotence" gets confused with "having IF NOT EXISTS in the CREATE." But idempotence is a property of the whole migration: every operation has to be safe to re-run.

How to tell: read your migration. For each operation, ask: "if this runs twice, does the second time work or fail?" If any of them fails, it isn't idempotent.

How to fix it: apply IF EXISTS / IF NOT EXISTS where they apply. For operations that don't support IF (RENAME, ALTER COLUMN), use DO $$ blocks with checks.

Mistake 3 (operational): not having a runbook until the first incident

Symptom: the first hung-migration incident is chaotic. The team improvises. It takes 4 hours to resolve. The postmortem says "we need a runbook" but nobody writes it until the next incident.

Why it happens: runbooks are "invisible" work in good times. They only get valued when they're needed. Roadmap pressure postpones them.

How to tell: does your team have a RUNBOOK-MIGRATION.md (or equivalent)? If it doesn't exist, you're vulnerable.

How to fix it: copy this capsule's template into your repo TODAY, before the next incident. Adapt it to your stack and table names.

Mistake 4 (conceptual): underestimating idle_in_transaction_session_timeout

Symptom: migrations sometimes hang not because the operation takes long, but because somebody (a dev, a script, a job) left an idle transaction open on the table.

Why it happens: without idle_in_transaction_session_timeout configured, an idle transaction can hold locks indefinitely. That parameter kills transactions that are "open but not doing anything" after a while.

How to tell: review your PostgreSQL config (or Alembic's env.py). Is idle_in_transaction_session_timeout set? If it's 0 (the default), idle transactions can live forever.

How to fix it: set idle_in_transaction_session_timeout = 60000 (1 minute) or similar at the DB or connection level. Legitimate transactions aren't idle for more than a few seconds. Prolonged idle ones are bugs.

Mistake 5 (operational): confusing pg_cancel_backend with pg_terminate_backend

Symptom: the dev runs pg_cancel_backend expecting to kill the backend, but the backend only cancels the current query and stays active. The transaction is still hung.

Why it happens: the names are subtly different:

• pg_cancel_backend(pid): cancels the current query, but leaves the transaction and the connection alive. If the backend is idle in transaction, it does nothing visible (there's no query to cancel).
• pg_terminate_backend(pid): terminates the whole connection. It rolls back the transaction and releases locks.

How to tell: if the backend is state = 'active', pg_cancel_backend cancels the query and that transaction can release locks. If it's state = 'idle in transaction', you need pg_terminate_backend.

How to fix it: memorize the difference. In the "hung migration" runbook, use the right command depending on the holder's state.

Exercises

Exercise 1: convert a non-idempotent migration into an idempotent one

This migration fails when it runs twice. Make it idempotent.

# alembic/versions/non_idempotent.py
def upgrade():
    op.add_column("tasks", sa.Column("status", sa.String(20), nullable=True))
    op.create_index("idx_tasks_status", "tasks", ["status"])
    op.execute("ALTER TABLE tasks ADD CONSTRAINT tasks_status_check CHECK (status IN ('open', 'closed'))")


def downgrade():
    op.execute("ALTER TABLE tasks DROP CONSTRAINT tasks_status_check")
    op.drop_index("idx_tasks_status", "tasks")
    op.drop_column("tasks", "status")

def upgrade():
    op.execute("SET lock_timeout = '5s'")

    # An idempotent ADD COLUMN
    op.execute("ALTER TABLE tasks ADD COLUMN IF NOT EXISTS status VARCHAR(20) NULL")

    # An idempotent CREATE INDEX
    op.execute("CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks (status)")

    # An idempotent ADD CONSTRAINT (it needs DO $$ because there's no IF NOT EXISTS for a constraint)
    op.execute(
        """
        DO $$
        BEGIN
            IF NOT EXISTS (
                SELECT 1 FROM pg_constraint WHERE conname = 'tasks_status_check'
            ) THEN
                ALTER TABLE tasks ADD CONSTRAINT tasks_status_check
                    CHECK (status IN ('open', 'closed'));
            END IF;
        END $$;
        """
    )


def downgrade():
    op.execute("SET lock_timeout = '5s'")

    # An idempotent DROP CONSTRAINT
    op.execute(
        "ALTER TABLE tasks DROP CONSTRAINT IF EXISTS tasks_status_check"
    )

    # An idempotent DROP INDEX
    op.execute("DROP INDEX IF EXISTS idx_tasks_status")

    # An idempotent DROP COLUMN
    op.execute("ALTER TABLE tasks DROP COLUMN IF EXISTS status")

Exercise 2: simulate an incident and apply the runbook

Create a local "hung migration" scenario:

1. Session A: open a transaction and take a lock on migration_test. Do NOT commit.
2. Session B: run alembic upgrade head with a migration that needs the lock. You're going to see it hang.
3. Session C ("yours", the incident response one): apply the runbook to diagnose and resolve.

Document the exact steps you ran and the outputs.

# alembic/versions/test_lock.py
def upgrade():
    op.execute("SET lock_timeout = '0'")  # No timeout, to reproduce the hang
    op.add_column("migration_test", sa.Column("test_col", sa.Integer(), nullable=True))

psql $DATABASE_URL

postgres=# BEGIN;
BEGIN
postgres=# LOCK TABLE migration_test IN ACCESS EXCLUSIVE MODE;
LOCK TABLE
-- Keep it like this, no commit/rollback

alembic upgrade head

psql $DATABASE_URL

postgres=# SELECT
    pg_class.relname AS table_name,
    pg_locks.mode,
    pg_locks.granted,
    pg_stat_activity.pid,
    pg_stat_activity.usename,
    pg_stat_activity.application_name,
    pg_stat_activity.state,
    age(now(), pg_stat_activity.query_start) AS duration,
    pg_stat_activity.query
FROM pg_locks
JOIN pg_class ON pg_locks.relation = pg_class.oid
JOIN pg_stat_activity ON pg_locks.pid = pg_stat_activity.pid
WHERE pg_class.relname = 'migration_test'
ORDER BY pg_locks.granted DESC;

   table_name  |       mode         | granted | pid  | usename  | application_name |       state         | duration |              query
---------------+--------------------+---------+------+----------+------------------+---------------------+----------+----------------------------------
 migration_test| ACCESS EXCLUSIVE   | t       | 1001 | postgres | psql             | idle in transaction | 00:03:15 | LOCK TABLE migration_test IN...
 migration_test| ACCESS EXCLUSIVE   | f       | 1002 | postgres | alembic          | active              | 00:02:50 | ALTER TABLE migration_test ADD...

postgres=# SELECT pg_terminate_backend(1001);
 pg_terminate_backend
----------------------
 t

INFO  [alembic.runtime.migration] Running upgrade ... -> test_lock, ...

postgres=# SELECT 1;
FATAL:  terminating connection due to administrator command
server closed the connection unexpectedly

Exercise 3: configure your app's env.py

Take the env.py template that comes with alembic init and modify it for production following this capsule's patterns. Include:

1. DATABASE_URL from an env var.
2. NullPool.
3. Connection-level timeouts (lock_timeout, statement_timeout, idle_in_transaction_session_timeout).
4. compare_type and compare_server_default enabled.
5. Multi-environment handling (dev/staging/prod).

# alembic/env.py
"""Alembic environment configuration for production multi-environment.

Environment variables:
    APP_ENV: dev | staging | prod (default: dev)
    DATABASE_URL: connection string (required for non-default env)
"""
import os
from logging.config import fileConfig

from sqlalchemy import engine_from_config, pool
from alembic import context

from app.db.models import Base


config = context.config


# ─── Multi-environment configuration ───
APP_ENV = os.environ.get("APP_ENV", "dev")

ENV_CONFIG = {
    "dev": {
        "default_database_url": "postgresql://localhost/myapp_dev",
        "lock_timeout_ms": 30_000,       # Permissive in dev
        "statement_timeout_ms": 0,        # No limit in dev
        "idle_timeout_ms": 60_000,        # 1 min
    },
    "staging": {
        "default_database_url": None,     # Required from env
        "lock_timeout_ms": 5_000,
        "statement_timeout_ms": 120_000,  # 2 min
        "idle_timeout_ms": 30_000,
    },
    "prod": {
        "default_database_url": None,     # Required from env
        "lock_timeout_ms": 5_000,
        "statement_timeout_ms": 60_000,
        "idle_timeout_ms": 10_000,
    },
}

if APP_ENV not in ENV_CONFIG:
    raise ValueError(
        f"Invalid APP_ENV: {APP_ENV}. "
        f"Expected one of: {list(ENV_CONFIG.keys())}"
    )

env_cfg = ENV_CONFIG[APP_ENV]

DATABASE_URL = os.environ.get("DATABASE_URL", env_cfg["default_database_url"])
if not DATABASE_URL:
    raise RuntimeError(
        f"DATABASE_URL is not set and there is no default for APP_ENV={APP_ENV}"
    )

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

# Logging
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# The target metadata for autogenerate
target_metadata = Base.metadata


def run_migrations_offline() -> None:
    """Generates SQL without connecting to the DB (to review migrations before applying)."""
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
        compare_type=True,
        compare_server_default=True,
    )

    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online() -> None:
    """Connects to the DB and applies migrations."""
    # Build the options for connection-level timeouts
    options_str = (
        f"-c lock_timeout={env_cfg['lock_timeout_ms']} "
        f"-c statement_timeout={env_cfg['statement_timeout_ms']} "
        f"-c idle_in_transaction_session_timeout={env_cfg['idle_timeout_ms']}"
    )

    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,  # Don't reuse connections
        connect_args={"options": options_str},
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            transaction_per_migration=True,
            compare_type=True,                # Detects type changes
            compare_server_default=True,      # Detects DEFAULT changes
        )

        # Log which environment and config we're using
        print(
            f"[alembic env] APP_ENV={APP_ENV}, "
            f"lock_timeout={env_cfg['lock_timeout_ms']}ms, "
            f"statement_timeout={env_cfg['statement_timeout_ms']}ms"
        )

        with context.begin_transaction():
            context.run_migrations()


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

# Dev
APP_ENV=dev alembic upgrade head

# Staging
APP_ENV=staging DATABASE_URL=$STAGING_URL alembic upgrade head

# Prod
APP_ENV=prod DATABASE_URL=$PROD_URL alembic upgrade head

Exercise 4: write a standalone backfill script

For your FastAPI app with the tasks table, write a standalone script that backfills tasks.priority = 0 following the capsule's pattern. It must:

1. Read DATABASE_URL from env.
2. Support --dry-run, --batch-size, --sleep as CLI args.
3. Log progress.
4. Verify at the end that no rows with NULL remain.
5. Be idempotent and retryable.

# scripts/backfills/001_backfill_tasks_priority.py
"""Backfill tasks.priority = 0 for rows with NULL.

Usage:
    python scripts/backfills/001_backfill_tasks_priority.py [options]

Options:
    --batch-size INT    Rows per batch (default: 10000)
    --sleep FLOAT       Seconds between batches (default: 0.1)
    --dry-run           Don't run UPDATEs, just count
    --max-id INT        Override max_id (to retry from a specific ID)
"""
from __future__ import annotations

import argparse
import logging
import os
import sys
import time
from dataclasses import dataclass

import psycopg2
from psycopg2.extensions import connection as PgConnection


logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)
log = logging.getLogger(__name__)


@dataclass
class BackfillConfig:
    database_url: str
    batch_size: int = 10_000
    sleep_sec: float = 0.1
    dry_run: bool = False
    max_id_override: int | None = None


@dataclass
class BackfillStats:
    batches: int = 0
    rows_updated: int = 0
    null_remaining: int | None = None


def setup_session(conn: PgConnection) -> None:
    """Configures timeouts on the session."""
    with conn.cursor() as cur:
        cur.execute("SET lock_timeout = '5s'")
        cur.execute("SET statement_timeout = '0'")  # No total timeout
    conn.commit()


def get_max_id(conn: PgConnection) -> int:
    """Gets the maximum ID of rows with priority NULL."""
    with conn.cursor() as cur:
        cur.execute(
            "SELECT COALESCE(MAX(id), 0) FROM tasks WHERE priority IS NULL"
        )
        return cur.fetchone()[0] or 0


def get_null_count(conn: PgConnection) -> int:
    """Counts rows with priority NULL."""
    with conn.cursor() as cur:
        cur.execute("SELECT COUNT(*) FROM tasks WHERE priority IS NULL")
        return cur.fetchone()[0]


def backfill_batch(
    conn: PgConnection,
    batch_min: int,
    batch_max: int,
    dry_run: bool,
) -> int:
    """Backfills a batch. Returns the number of rows updated."""
    with conn.cursor() as cur:
        if dry_run:
            cur.execute(
                """
                SELECT COUNT(*) FROM tasks
                WHERE id BETWEEN %s AND %s AND priority IS NULL
                """,
                (batch_min, batch_max),
            )
            return cur.fetchone()[0]
        cur.execute(
            """
            UPDATE tasks SET priority = 0
            WHERE id BETWEEN %s AND %s AND priority IS NULL
            """,
            (batch_min, batch_max),
        )
        return cur.rowcount or 0


def run_backfill(cfg: BackfillConfig) -> BackfillStats:
    stats = BackfillStats()

    conn = psycopg2.connect(cfg.database_url)
    conn.autocommit = False

    try:
        setup_session(conn)

        max_id = cfg.max_id_override or get_max_id(conn)
        if max_id == 0:
            log.info("Nothing to backfill (max_id = 0)")
            return stats

        log.info(
            f"Backfill started: max_id={max_id}, batch_size={cfg.batch_size}, "
            f"dry_run={cfg.dry_run}"
        )

        batch_min = 0
        while batch_min <= max_id:
            batch_max = batch_min + cfg.batch_size - 1

            try:
                rows_in_batch = backfill_batch(
                    conn, batch_min, batch_max, cfg.dry_run
                )
                if not cfg.dry_run:
                    conn.commit()

                stats.rows_updated += rows_in_batch
                stats.batches += 1

                if stats.batches % 10 == 0:
                    pct = min(100, (batch_min / max_id) * 100)
                    log.info(
                        f"Batch {stats.batches}: progress {pct:.1f}% "
                        f"(total rows: {stats.rows_updated})"
                    )
            except psycopg2.Error as e:
                conn.rollback()
                log.error(
                    f"Error in batch {batch_min}-{batch_max}: {e}. "
                    f"Skipping this batch."
                )
                # Continue with the next batch (idempotence saves us)

            batch_min = batch_max + 1
            time.sleep(cfg.sleep_sec)

        # Final verification
        stats.null_remaining = get_null_count(conn)
        log.info(
            f"Backfill complete. Batches: {stats.batches}, "
            f"rows updated: {stats.rows_updated}, "
            f"rows still NULL: {stats.null_remaining}"
        )

        if stats.null_remaining > 0 and not cfg.dry_run:
            log.warning(
                f"WARNING: there are still {stats.null_remaining} rows with priority IS NULL. "
                "Possible cause: the old app is still inserting NULLs. "
                "Check the app deploy and re-run."
            )

        return stats
    finally:
        conn.close()


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--batch-size", type=int, default=10_000)
    parser.add_argument("--sleep", type=float, default=0.1)
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument("--max-id", type=int, default=None)
    args = parser.parse_args()

    database_url = os.environ.get("DATABASE_URL")
    if not database_url:
        log.error("DATABASE_URL is not set")
        return 1

    cfg = BackfillConfig(
        database_url=database_url,
        batch_size=args.batch_size,
        sleep_sec=args.sleep,
        dry_run=args.dry_run,
        max_id_override=args.max_id,
    )

    stats = run_backfill(cfg)

    if stats.null_remaining and stats.null_remaining > 0:
        return 1  # Incomplete backfill
    return 0


if __name__ == "__main__":
    sys.exit(main())

Exercise 5: write your own adapted runbook

Take this capsule's RUNBOOK-MIGRATION.md template and adapt it to your app. Specifically:

1. Replace the generic table names with your app's.
2. Add your teams'/Slack channels' names for escalation.
3. Include the exact commands to connect to your production DB.
4. Add any step specific to your stack (e.g. the Datadog/Grafana command to check metrics, links to dashboards).

Share it with your team and ask them for feedback.

## Escalation

If you haven't resolved it after 30 minutes:
- Notify in the `#platform-incidents` Slack channel
- Tag @on-call-dba (the PagerDuty rotation)
- Create a Jira ticket with the `incident-db` label

## Connecting to production

\`\`\`bash
# Via a bastion host (no direct exposure)
ssh bastion-prod "psql $PROD_DATABASE_URL"

# Via the cloudsql proxy if you use GCP
cloud_sql_proxy -instances=myproject:us-east1:prod=tcp:5432 &
psql "host=localhost port=5432 dbname=myapp user=migrate_user"
\`\`\`

## Relevant dashboards

- [DB Metrics — Lock Waits](https://grafana.example.com/d/db-locks)
- [App Latency — by Endpoint](https://grafana.example.com/d/app-latency)
- [Active Connections — by App](https://grafana.example.com/d/db-connections)

## Metrics to monitor during an incident

- `pg_stat_activity` count by state
- p99 latency of `/api/tasks/*`
- 5xx error rate at the API gateway

Summary and next step

In this capsule you learned:

• Migrations are explicit operations separate from the code deploy. Don't run them at container startup.
• Idempotence is a property of the whole migration, not just of individual operations. IF NOT EXISTS, IF EXISTS, and DO $$ blocks for cases with no native IF.
• env.py configures Alembic's operational foundation in production: NullPool, connection-level timeouts, compare_type, multi-environment.
• The RUNBOOK-MIGRATION.md is the most important asset for incident response. Without a runbook, incidents are chaos. With a runbook, they're routine.
• pg_cancel_backend cancels the query, pg_terminate_backend terminates the connection — they're different. Memorize the difference.
• Backfills as standalone scripts let you run them at a convenient time without blocking deploys.
• idle_in_transaction_session_timeout is the parameter that quietly prevents the most incidents: it kills zombie transactions before they hold locks.

Before moving on you should be able to:

• Configure env.py with all the operational patterns (NullPool, timeouts, multi-environment).
• Write idempotent migrations that survive retries.
• Create standalone backfill scripts with a CLI, dry-run, and verification.
• Apply the hung-migration runbook under pressure, knowing the difference between pg_cancel_backend and pg_terminate_backend.
• Adapt the runbook to your specific app with concrete commands and links.
• Decide between the 3 orchestration options (a CI/CD step, a manual job, a K8s Job) according to the team's context.

Next capsule — Blue-green and database rollback strategies. You now have the complete production operation: individual techniques (capsules 02-05) + a production workflow (capsule 06). Capsule 07 closes the module covering two critical topics we haven't touched yet: when blue-green with a DB actually works (spoiler: only with additive-only changes) and how to plan the rollback window at each phase of expand-contract. It's the capsule that gives you the realistic limits of the techniques: knowing what does NOT work saves you from adopting solutions that look magical but break.

Resources

1. Alembic — env.py reference — complete documentation of the env.py file.
2. PostgreSQL — pg_cancel_backend / pg_terminate_backend — the differences and the exact semantics.
3. PostgreSQL — Monitoring lock waits — ready-made queries to diagnose locks.
4. GitLab — Database migrations runbook — a runbook reference from an experienced team.
5. Heroku — Postgres lock monitoring — a specific analysis of detecting hung locks.
6. Datadog — PostgreSQL monitoring — which metrics to watch to detect migration incidents early.
7. Squawk linter — a linter to detect dangerous patterns in SQL migrations before the deploy.

Module 5 — SQL Patterns for Production APIs Guide

Next capsule: Blue-green and database rollback strategies — when blue-green works, when it doesn't, and how to plan realistic rollback windows.