Running Alembic migrations in CI/CD pipelines

Run migrations in a dedicated pre-deploy job holding a PostgreSQL advisory lock, not at application start-up — with several replicas starting at once, start-up migrations race, and the loser either fails or applies the same DDL twice. This is the deployment half of managing multiple heads and database branching.

Quick Answer

Before — migrating at application start-up:

@app.on_event("startup")
async def migrate() -> None:
    await asyncio.to_thread(command.upgrade, config, "head")   # every replica does this

After — a dedicated job holding an advisory lock:

"""deploy/migrate.py — run once per deploy, before the new replicas start."""
from __future__ import annotations

import asyncio
import os
import sys

from alembic import command
from alembic.config import Config
from alembic.script import ScriptDirectory
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine

# Any stable 64-bit integer; it names this application's migration lock.
MIGRATION_LOCK_KEY = 0x5A1CE_M1G8 & 0x7FFFFFFFFFFFFFFF


async def main() -> int:
    url = os.environ["DATABASE_URL"]
    config = Config("alembic.ini")
    config.set_main_option("sqlalchemy.url", url.replace("asyncpg", "psycopg2"))

    heads = ScriptDirectory.from_config(config).get_heads()
    if len(heads) != 1:
        print(f"refusing to migrate: {len(heads)} heads ({', '.join(heads)})",
              file=sys.stderr)
        return 2

    engine = create_async_engine(url, poolclass=None)
    try:
        async with engine.connect() as connection:
            # Session-level: released automatically if this job is killed.
            await connection.execute(
                text("SELECT pg_advisory_lock(:key)"), {"key": MIGRATION_LOCK_KEY}
            )
            await asyncio.to_thread(command.upgrade, config, "head")
    finally:
        await engine.dispose()
    return 0


if __name__ == "__main__":
    raise SystemExit(asyncio.run(main()))

The job exits non-zero on failure, which is what makes the rollout stop rather than proceeding with replicas that expect a schema the database does not have.

Execution Context & Async Workflow Integration

Alembic's command API is synchronous even when env.py drives an async engine, so the job above runs command.upgrade in a thread. That is not a workaround — the migration genuinely is a blocking operation with nothing else to interleave with — but it does mean the advisory lock has to be held on a different connection from the one Alembic uses, which is exactly what the code above does.

The lock key deserves a moment's thought. pg_advisory_lock takes a 64-bit integer and there is no registry, so two applications sharing a database can collide by accident. Deriving the key from the application name removes that:

import hashlib

def lock_key(application: str) -> int:
    digest = hashlib.sha256(application.encode()).digest()[:8]
    return int.from_bytes(digest, "big", signed=True)


MIGRATION_LOCK_KEY = lock_key("orders-api")

For the case where a dedicated job genuinely is not available — a platform that only runs containers, with no pre-deploy hook — the same lock makes start-up migrations safe:

# Async — start-up migration that is safe with N replicas
async def migrate_on_startup(url: str) -> None:
    engine = create_async_engine(url, poolclass=None)
    try:
        async with engine.connect() as connection:
            # try_advisory_lock returns immediately: whoever gets it migrates,
            # everyone else proceeds to start up and serve.
            got = await connection.scalar(
                text("SELECT pg_try_advisory_lock(:key)"), {"key": MIGRATION_LOCK_KEY}
            )
            if not got:
                return
            await asyncio.to_thread(command.upgrade, config, "head")
    finally:
        await engine.dispose()

pg_try_advisory_lock rather than pg_advisory_lock is deliberate: the replicas that do not get the lock should start serving rather than blocking behind the migration. The weakness compared with a dedicated job remains — a failure is logged by one replica among many and does not stop the rollout — which is why this is the fallback rather than the recommendation.

Resolving Warnings, Errors & Common Mistakes

Advanced: Reviewable SQL and a Gate for Risky Operations

For any environment where a DBA reviews changes, Alembic's offline mode prints the SQL without running it — and emitting that into the build log gives an auditable record of exactly what a deploy was about to do.

# Emit the SQL for the range being applied, without connecting
$ alembic upgrade c3a1029ef455:head --sql | tee migration.sql

That output is also the right input for an automated risk gate. Certain operations should never appear in a deploy that happens without a human present:

"""Fail the pipeline when a migration contains an operation that needs a person."""
from __future__ import annotations

import re
import sys

RISKY = {
    r"\bDROP\s+TABLE\b": "drops a table",
    r"\bDROP\s+COLUMN\b": "drops a column",
    r"\bALTER\s+COLUMN\b.*\bTYPE\b": "changes a column type",
    r"\bCREATE\s+INDEX\b(?!\s+CONCURRENTLY)": "builds an index without CONCURRENTLY",
    r"\bUPDATE\b(?!.*\bWHERE\b)": "updates every row of a table",
}


def check(sql: str) -> int:
    findings = [
        why for pattern, why in RISKY.items()
        if re.search(pattern, sql, re.IGNORECASE | re.DOTALL)
    ]
    for why in findings:
        print(f"BLOCKED: this migration {why}", file=sys.stderr)
    return 1 if findings else 0


if __name__ == "__main__":
    raise SystemExit(check(sys.stdin.read()))

The point is not that these operations are forbidden — sometimes a column really does need dropping — but that they should require an explicit acknowledgement rather than riding along in a routine deploy. A label on the pull request, or an environment variable set deliberately in a manual run, is enough to bypass the gate with a record of who decided.

The CREATE INDEX check is the one that earns its keep most often. An index built without CONCURRENTLY takes a lock that blocks writes for the duration, which on a large table is an outage nobody planned — and the fix, covered in zero-downtime schema migration strategies, also requires transaction_per_migration in env.py, so the gate catches two related mistakes at once.

When the Migration Job Fails

A failed migration job is a good outcome compared with the alternatives, because the rollout stops and the old replicas keep serving against the old schema. What matters next is knowing which state the database is in.

With the default single-transaction behaviour, a failure rolls the entire run back and alembic_version is unchanged — the database is exactly where it started, and the fix is to correct the revision and redeploy.

With transaction_per_migration=True, revisions before the failing one have committed, so the database is at an intermediate revision. alembic current tells you which. From there the choices are to fix the failing revision and re-run, which resumes from where it stopped, or to add a new revision that corrects the situation. Downgrading is rarely the right answer, for the reasons in [rolling back a failed Alembic migration in production](' + P + 'rolling-back-a-failed-alembic-migration-in-production/).

Two habits make the recovery quicker. Have the job log alembic current both before and after, so the build output records exactly where the database started and finished — this removes an entire round of "which environment is at which revision" during an incident. And set a lock_timeout on the migration connection:

await connection.execute(text("SET lock_timeout = '10s'"))

Without it, a migration needing a table lock waits behind whatever long-running transaction happens to exist — an open psql session, a slow report — and the deploy hangs rather than failing. With it, the job fails in ten seconds with a message naming the lock, which is a diagnosis rather than a mystery.

Frequently Asked Questions

Related

• Managing Multiple Heads and Database Branching — The parent guide: the single-head rule this pipeline enforces.
• Resolving Alembic multiple head revisions errors — The error the pipeline check is designed to catch before a deploy.
• Rolling back a failed Alembic migration in production — What to do after the migration job exits non-zero.
• Zero-downtime schema migration strategies — Why migrations must be compatible with the version still running during the rollout.