Database Migrations & Schema Evolution

Step 1: Install and Initialize Alembic

# Install Alembic (skip if the setup block's virtualenv is still active).
# Python packages belong in a virtualenv. On Debian/Ubuntu a bare
# "pip install" is refused (PEP 668: externally-managed-environment).
python3 -m venv .venv && source .venv/bin/activate
pip install alembic sqlalchemy "psycopg[binary]"

# Initialize Alembic in your project
alembic init migrations

# This creates alembic.ini in the CURRENT directory, alongside (not
# inside) the migrations package:
#
# .
# ├── alembic.ini           # Configuration file: lives in the project root
# └── migrations/
#     ├── env.py            # Environment setup, runs on every command
#     ├── README
#     ├── script.py.mako    # Template new migrations are rendered from
#     └── versions/         # Migration files go here

Creating directory /path/to/project/migrations ...  done
Creating directory /path/to/project/migrations/versions ...  done
Generating /path/to/project/migrations/script.py.mako ...  done
Generating /path/to/project/alembic.ini ...  done
Generating /path/to/project/migrations/README ...  done
Generating /path/to/project/migrations/env.py ...  done
Please edit configuration/connection/logging settings in
/path/to/project/alembic.ini before proceeding.

Note where alembic.ini lands: in the directory you ran the command from, beside migrations/ rather than inside it. Every subsequent alembic command has to run from that same directory, or it will not find its configuration.

Step 2: Configure Database Connection

# alembic.ini - update the database URL
#
# Use postgresql+psycopg://, NOT postgresql://. The bare "postgresql://"
# scheme still resolves to psycopg2, which is not what the setup block
# installed, and Alembic dies with:
#     ModuleNotFoundError: No module named 'psycopg2'
sqlalchemy.url = postgresql+psycopg://demo:demo@localhost:5432/demo

The driver prefix is not cosmetic. SQLAlchemy still maps the bare postgresql:// scheme to psycopg2, so with only psycopg[binary] (psycopg 3) installed, as the setup block above installs it, every Alembic command dies with ModuleNotFoundError: No module named 'psycopg2'. Writing postgresql+psycopg:// selects psycopg 3 explicitly.

# migrations/env.py - point Alembic at your models, and read the URL
# from the environment so CI and production never hardcode credentials.
import os

from myapp.models import Base   # your SQLAlchemy Base

# Nothing reads DATABASE_URL unless you write this: the stock env.py only
# ever looks at sqlalchemy.url in alembic.ini. Setting the variable in a
# CI job without these two lines has no effect at all.
if os.getenv("DATABASE_URL"):
    config.set_main_option("sqlalchemy.url", os.getenv("DATABASE_URL"))

target_metadata = Base.metadata   # lets --autogenerate detect model changes

Two jobs get done in env.py. Assigning target_metadata is what makes --autogenerate able to compare your models against the live schema; leave it as None and autogeneration silently produces empty migrations. The DATABASE_URL block matters just as much in CI: nothing reads that variable unless you wire it up, so a pipeline that sets it without these lines quietly runs every migration against whatever URL is sitting in alembic.ini.

Step 3: Create Your First Migration

# Define SQLAlchemy model
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.orm import declarative_base
from datetime import datetime, timezone

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(100), nullable=False)
    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))

# Then, from a shell in the directory holding alembic.ini:
#   alembic revision --autogenerate -m "Create users table"

INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.autogenerate.compare.tables] Detected added table 'users'
Generating /path/to/project/migrations/versions/
  d2bb6019fab3_create_users_table.py ...  done

The revision id (d2bb6019fab3 here) is randomly generated, so yours will differ. That hex string, not the filename or the date, is what Alembic uses to order migrations and record which ones have run.

Step 4: Review Generated Migration

# migrations/versions/d2bb6019fab3_create_users_table.py
"""Create users table

Revision ID: d2bb6019fab3
Revises:
Create Date: 2026-08-01 12:05:36.495803

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision: str = 'd2bb6019fab3'
down_revision: Union[str, Sequence[str], None] = None   # First migration
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    """Upgrade schema."""
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('users',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('username', sa.String(length=50), nullable=False),
    sa.Column('email', sa.String(length=100), nullable=False),
    sa.Column('created_at', sa.DateTime(), nullable=True),
    sa.PrimaryKeyConstraint('id'),
    sa.UniqueConstraint('username')
    )
    # ### end Alembic commands ###


def downgrade() -> None:
    """Downgrade schema."""
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_table('users')
    # ### end Alembic commands ###

Step 5: Apply Migration to Database

# Apply all pending migrations
alembic upgrade head

# Then confirm where the database now stands
alembic current

INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade  -> d2bb6019fab3, Create users table

INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
d2bb6019fab3 (head)

The database now has the users table, and Alembic has recorded which revision it is on in a table of its own: SELECT * FROM alembic_version; returns a single row holding d2bb6019fab3. That one row is the entire state Alembic keeps, which is why restoring a database backup without also checking out the matching code can leave the two out of step, and why nobody should ever edit that table by hand.

Pattern: Expand/Contract (Multi-Phase Deploy)

Example: Adding a Column (Zero-Downtime)

# migrations/versions/phase1_add_full_name.py
# Phase 1: add the column, nullable and with no default, which is a
# catalog-only change: no table rewrite, lock held for microseconds.
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa

revision: str = 'phase1'
down_revision: Union[str, Sequence[str], None] = 'd2bb6019fab3'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    op.add_column('users', sa.Column('full_name', sa.String(200), nullable=True))


def downgrade() -> None:
    op.drop_column('users', 'full_name')

# Create the file with "alembic revision -m ..." and apply it with
# "alembic upgrade head", both from a shell, not from inside upgrade().

Adding a nullable column with no default is a catalog-only change: PostgreSQL records it in pg_attribute and touches no rows, so the lock is held for microseconds regardless of table size. The application model ships in the same deploy, and the important part is that it keeps the column nullable and tolerates None:

# myapp/models.py - the application model, updated in the same deploy.
# full_name is nullable here because Phase 2 has not backfilled yet, and
# old rows still hold NULL. Tightening this too early is what breaks the
# rollout.
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.orm import declarative_base
from datetime import datetime, timezone

Base = declarative_base()


class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(100), nullable=False)
    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
    full_name = Column(String(200), nullable=True)   # New column

    @property
    def name_for_display(self):
        return self.full_name or self.username       # Fall back to the old one

Old instances still running the previous release keep working because they simply never mention full_name. New instances write it on every create and update, and read through name_for_display, which falls back to username for the rows nobody has backfilled yet. Both versions run against the same schema at the same time, which is the whole point of the expand phase.

Phase 2: Backfill Existing Data

# Phase 2: backfill in batches. Every batch is its own short
# transaction, so nothing holds a lock long enough to block traffic.
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from myapp.models import User   # the model from the previous block

engine = create_engine("postgresql+psycopg://demo:demo@localhost:5432/demo")
Session = sessionmaker(bind=engine)


def backfill_full_names(batch_size=20000):
    """Populate full_name for existing users."""
    session = Session()
    total = 0
    while True:
        # .is_(None) rather than "== None": both compile to IS NULL, but
        # the operator form trips linters (E711) and breaks outright if
        # anyone "simplifies" it to "not User.full_name".
        users = session.query(User).filter(
            User.full_name.is_(None)
        ).limit(batch_size).all()

        if not users:
            break   # Nothing left to backfill

        # Real deployments backfill from the source of truth (an import
        # file, another table); here we derive a placeholder from the
        # username, since that is all this table has.
        for user in users:
            user.full_name = user.username.replace("user", "User ")

        session.commit()
        total += len(users)
        print(f"Backfilled {total} users...")

    print("Backfill complete!")


# Safe to run during business hours: no long locks, no downtime.
backfill_full_names()

Backfilled 20000 users...
Backfilled 40000 users...
Backfilled 60000 users...
Backfilled 80000 users...
Backfilled 100000 users...
Backfill complete!

Run against 100,000 rows, this took about four seconds and never blocked anything: each batch is a separate short transaction, so the row locks it takes are released five times over the run instead of being held from start to finish. That is the difference from a single UPDATE users SET ..., which would hold locks on every row it had touched until the whole statement committed. Batch size is a tradeoff, not a constant: larger batches finish sooner, smaller ones hold fewer locks at once and leave more room for production traffic.

Phase 3: Make Column NOT NULL (Safe Now!)

# migrations/versions/phase3_full_name_not_null.py
# Phase 3: enforce NOT NULL, safe now that every row has a value.
#
# The direct route, "ALTER COLUMN full_name SET NOT NULL", holds ACCESS
# EXCLUSIVE for as long as it takes to scan the table and prove no row is
# NULL. Every read and write waits for that whole scan.
#
# The route below splits the work so the SCAN happens under a weaker lock:
# ADD CONSTRAINT ... NOT VALID records the rule without checking anything,
# VALIDATE CONSTRAINT does the scan under SHARE UPDATE EXCLUSIVE (which
# does not block reads or writes), and SET NOT NULL then reuses the
# already-validated constraint instead of rescanning.
from alembic import op


def upgrade() -> None:
    op.execute(
        "ALTER TABLE users ADD CONSTRAINT full_name_not_null "
        "CHECK (full_name IS NOT NULL) NOT VALID"
    )
    op.execute("ALTER TABLE users VALIDATE CONSTRAINT full_name_not_null")
    op.execute("ALTER TABLE users ALTER COLUMN full_name SET NOT NULL")
    # Redundant once the column itself is NOT NULL.
    op.execute("ALTER TABLE users DROP CONSTRAINT full_name_not_null")


def downgrade() -> None:
    op.execute("ALTER TABLE users ALTER COLUMN full_name DROP NOT NULL")

# shell: alembic revision -m "Phase 3: Make full_name NOT NULL"
# shell: alembic upgrade head

-- Direct route, on a fully backfilled 100,000-row table:
ALTER TABLE users ALTER COLUMN full_name SET NOT NULL;
Time: 6.920 ms      <- entire table blocked for this long

-- Split route, same table:
ALTER TABLE ... ADD CONSTRAINT ... NOT VALID;
Time: 1.955 ms      <- ACCESS EXCLUSIVE, but no scan
ALTER TABLE users VALIDATE CONSTRAINT fnn;
Time: 6.995 ms      <- the scan, under SHARE UPDATE EXCLUSIVE
ALTER TABLE users ALTER COLUMN full_name SET NOT NULL;
Time: 1.241 ms      <- reuses the validated constraint, no rescan
ALTER TABLE users DROP CONSTRAINT fnn;
Time: 1.765 ms

Read those numbers carefully, because they do not say what people expect. The split route takes more wall-clock time in total (about 12 ms against 6.9 ms), and at 100,000 rows the direct route is over in a few milliseconds anyway. The win is not speed, it is which lock is held during the expensive part: the scan happens inside VALIDATE CONSTRAINT under SHARE UPDATE EXCLUSIVE, and a concurrent UPDATE against the table succeeds while it runs, which is verifiable with the pg_locks query from the top of this lesson. Scale the table to a billion rows and that scan becomes minutes: minutes of ordinary traffic under the split route, minutes of total outage under the direct one.

Rolling Back a Migration

# Where is the database now?
alembic current

# Step back exactly one revision
alembic downgrade -1

# Or land on a specific revision
alembic downgrade d2bb6019fab3

# Or unwind everything. This runs every downgrade() in reverse and leaves
# an empty schema: the only table still standing is alembic_version.
alembic downgrade base

INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
phase1 (head)

INFO  [alembic.runtime.migration] Running downgrade phase1 -> d2bb6019fab3, Phase 1: Add full_name column

INFO  [alembic.runtime.migration] Running downgrade phase1 -> d2bb6019fab3, Phase 1: Add full_name column
INFO  [alembic.runtime.migration] Running downgrade d2bb6019fab3 -> , Create users table

            List of relations
 Schema |      Name       | Type  | Owner
--------+-----------------+-------+-------
 public | alembic_version | table | demo
(1 row)

Note what downgrade base does not remove: alembic_version survives, now holding no row at all. That is deliberate, since Alembic still needs somewhere to record the next upgrade. It is also a reminder that a downgrade is only as good as the downgrade() functions someone bothered to write, and that they restore schema, never data: the op.drop_column above discards every full_name value permanently, and re-running the upgrade brings the column back empty.

Team Workflow Best Practices

CI/CD Integration

# .github/workflows/deploy.yml
name: Deploy with Migrations

on:
  push:
    branches: [main]

jobs:
  deploy:
    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

      # Order matters. "alembic check" compares your models against the
      # database and requires it to already be at head, so running it
      # BEFORE the upgrade fails with "Target database is not up to date".
      # Upgrade first, then check for model changes nobody generated a
      # migration for.
      - name: Apply migrations (staging)
        run: alembic upgrade head
        env:
          DATABASE_URL: ${{ secrets.STAGING_DB_URL }}

      - name: Check for un-migrated model changes
        run: alembic check
        env:
          DATABASE_URL: ${{ secrets.STAGING_DB_URL }}

      - name: Run tests
        run: pytest
        env:
          DATABASE_URL: ${{ secrets.STAGING_DB_URL }}

      - name: Apply migrations (production)
        if: success()
        run: alembic upgrade head
        env:
          DATABASE_URL: ${{ secrets.PROD_DB_URL }}

      - name: Deploy application
        run: ./deploy.sh

Two ordering details in there are worth stealing. alembic check runs after the upgrade, not before: it compares your models against the live schema and requires the database to already be at head, so putting it first fails with FAILED: Target database is not up to date on every run. What it actually catches is the common mistake of editing a model and forgetting to generate the migration, reporting No new upgrade operations detected. when things are clean. And every step that touches the database carries its own DATABASE_URL, which only works because env.py was wired to read it back in Step 2; without those two lines the variable is decoration and the job migrates whatever is hardcoded in alembic.ini.

Adding an Index Without Blocking Writes

A plain CREATE INDEX takes a lock that blocks every write to the table until the index is built, which on a large table means an outage. CONCURRENTLY avoids that by building the index in two passes while writes continue. It comes with one hard constraint that catches nearly everyone the first time.

# Building an index CONCURRENTLY does not block writes, which is the
# only way to add one to a busy table. But it cannot run inside a
# transaction, and Alembic wraps every migration in one, so the obvious
# version fails:
#
#     op.create_index('ix_users_full_name', 'users', ['full_name'],
#                     postgresql_concurrently=True)
#
#     sqlalchemy.exc.InternalError: (psycopg.errors.ActiveSqlTransaction)
#     CREATE INDEX CONCURRENTLY cannot run inside a transaction block
#
# autocommit_block() suspends the surrounding transaction for the
# duration, which is what makes it work.
from alembic import op


def upgrade() -> None:
    with op.get_context().autocommit_block():
        op.create_index('ix_users_full_name', 'users', ['full_name'],
                        postgresql_concurrently=True)


def downgrade() -> None:
    with op.get_context().autocommit_block():
        op.drop_index('ix_users_full_name', 'users', postgresql_concurrently=True)

INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade phase1 -> idxtest

                  List of relations
 Schema |        Name        | Type  | Owner | Table
--------+--------------------+-------+-------+-------
 public | ix_users_full_name | index | demo  | users
(1 row)

The tradeoff for not blocking writes is that a concurrent build can fail, typically on a unique index when existing rows violate it. When it does, it leaves an invalid index behind that still costs write overhead while being useless for queries. It will not clean itself up. Check for them with SELECT indexrelid::regclass FROM pg_index WHERE NOT indisvalid; and drop what turns up before retrying.