Alembic Migrations Interview Questions & Answers

Alembic is SQLAlchemy's official database migration tool. It tracks schema changes as versioned migration scripts and applies them incrementally.

Base.metadata.create_all() only creates missing tables — it cannot alter existing columns, add indexes, rename tables, or roll back changes. Alembic can do all of this and keeps an audit trail.

alembic upgrade head     # apply all pending migrations
alembic downgrade -1     # roll back the last migration
alembic history          # show migration history

Rule of thumb: use create_all only in local dev and test databases; use Alembic for every schema change in staging and production.

pip install alembic
alembic init alembic    # creates alembic/ directory and alembic.ini

Directory structure created:

alembic/
├── env.py          # migration environment (connect to DB)
├── script.py.mako  # template for new migration files
└── versions/       # generated migration scripts live here
alembic.ini         # config (sqlalchemy.url, etc.)

Edit alembic.ini to set the DB URL, or set it dynamically in env.py:

# alembic/env.py
from app.config import settings
config.set_main_option("sqlalchemy.url", settings.database_url)

Rule of thumb: set the DB URL from environment variables in env.py rather than hard-coding it in alembic.ini — alembic.ini is committed to git.

Point target_metadata to your Base.metadata so Alembic can compare the current schema against your ORM models:

# alembic/env.py
from app.db.base import Base          # import Base
from app.db.models import user, order  # ensure all models are imported!

target_metadata = Base.metadata

def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
    )
    with connectable.connect() as connection:
        context.configure(connection=connection, target_metadata=target_metadata)
        with context.begin_transaction():
            context.run_migrations()

The model imports are critical — Alembic only knows about tables whose models are imported at migration time.

Rule of thumb: create a db/base_all.py that imports every ORM model and import it in env.py — it's easy to forget a new model.

alembic revision --autogenerate -m "add users table"

Alembic compares target_metadata (your ORM models) to the current DB schema and generates a script with the diff:

# alembic/versions/abc123_add_users_table.py
def upgrade():
    op.create_table(
        "users",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("name", sa.String(100), nullable=False),
        sa.PrimaryKeyConstraint("id"),
    )

def downgrade():
    op.drop_table("users")

Always review the generated script before applying — Alembic can't detect renames (it sees a drop + create) and may miss custom indexes or constraints.

Rule of thumb: treat autogenerate as a starting point, not a final answer — always read the diff and edit it if needed before committing.

# Apply all pending migrations
alembic upgrade head

# Apply exactly N migrations forward
alembic upgrade +2

# Roll back the most recent migration
alembic downgrade -1

# Roll back to a specific revision
alembic downgrade abc123

# Roll back everything
alembic downgrade base

Alembic tracks the current revision in the alembic_version table in the DB.

Rule of thumb: always write a proper downgrade() function — even if you never plan to roll back in production, it's invaluable in development and CI.

Use AsyncEngine and run_async_migrations in env.py:

# alembic/env.py
import asyncio
from sqlalchemy.ext.asyncio import create_async_engine

def do_run_migrations(connection):
    context.configure(connection=connection, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()

async def run_async_migrations():
    engine = create_async_engine(settings.database_url)
    async with engine.connect() as conn:
        await conn.run_sync(do_run_migrations)
    await engine.dispose()

def run_migrations_online():
    asyncio.run(run_async_migrations())

The key is conn.run_sync(do_run_migrations) — Alembic's migration API is synchronous; you bridge it with run_sync.

Rule of thumb: async migration setup is boilerplate — copy the pattern from the official Alembic docs rather than hand-rolling it.

Never run migrations from inside the FastAPI lifespan function in production. Instead:

Option A — Pre-deployment step (recommended):

# In CI/CD pipeline, before rolling out new pods
alembic upgrade head

Option B — Init container (Kubernetes):

initContainers:
  - name: migrate
    image: myapp:latest
    command: ["alembic", "upgrade", "head"]

Reasons to avoid in-app migrations:

• Multiple pods start simultaneously → race condition running migrations twice.
• A failed migration crashes the app before it can serve requests.
• Slow migrations block the startup health check.

Rule of thumb: decouple migrations from app startup — run them as a separate step with exactly one executor, with a rollback plan if the migration fails.

Use op.execute() for SQL or op.get_bind() to access the connection for ORM-style operations:

from alembic import op
import sqlalchemy as sa

def upgrade():
    # Schema change
    op.add_column("users", sa.Column("full_name", sa.String(200)))

    # Data migration — populate full_name from existing columns
    conn = op.get_bind()
    conn.execute(
        sa.text("UPDATE users SET full_name = first_name || ' ' || last_name")
    )

    # Make it non-nullable now that data is populated
    op.alter_column("users", "full_name", nullable=False)

def downgrade():
    op.drop_column("users", "full_name")

Rule of thumb: do schema changes, then data migration, then add constraints — never add a NOT NULL constraint before populating the column.

A multiple heads situation occurs when two developers create migrations from the same base revision — Alembic has two "latest" revisions and doesn't know which to apply next.

alembic heads   # shows: abc123  xyz789  (two heads)
alembic merge heads -m "merge feature_a and feature_b"

This creates a new migration that has both revisions as parents, resolving the divergence. The merge migration itself is usually empty.

Rule of thumb: include the alembic_version table (or the versions/ directory) in code review — catch diverged heads early rather than after merging to main.

Alembic creates a single-row table alembic_version in the target database to track which revision is currently applied:

SELECT * FROM alembic_version;
-- version_num
-- abc123def456   (hash of the latest applied migration)

When you run alembic upgrade head, it reads this table, finds all unapplied revisions in order, runs their upgrade() functions, and updates version_num.

Rule of thumb: never edit alembic_version manually — if it gets out of sync, use alembic stamp <revision> to set it without running migration code.

alembic stamp <revision> sets the alembic_version table to the given revision without running any migration code. Use it when:

1. You're bootstrapping a new DB that already has the schema (e.g., cloned from prod) and you want to mark it as current without running all migrations.
2. You manually applied a schema change and want Alembic to know it's done.

alembic stamp head      # mark DB as fully migrated (no code runs)
alembic stamp abc123    # mark a specific revision as current

Rule of thumb: use alembic stamp head when creating a fresh DB from create_all to keep Alembic in sync; never use it to skip failed migrations in production.

Run upgrade + downgrade in CI against a test database:

# In CI
alembic upgrade head      # apply all migrations
alembic downgrade base    # roll all the way back
alembic upgrade head      # re-apply (round-trip test)

For individual migration testing in pytest:

@pytest.fixture
def migrated_db(postgres_url):
    engine = create_engine(postgres_url)
    with engine.begin() as conn:
        alembic_cfg = Config("alembic.ini")
        alembic_cfg.set_main_option("sqlalchemy.url", postgres_url)
        command.upgrade(alembic_cfg, "head")
    yield engine
    with engine.begin() as conn:
        command.downgrade(alembic_cfg, "base")

Rule of thumb: test up + down in CI on every PR — a migration that upgrades successfully but can't downgrade will strand you if a release needs to roll back.