Alembic Cheatsheet 08 — CI Integration

name: Migrations
on: [pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:17
        env:
          POSTGRES_PASSWORD: test
          POSTGRES_DB: test
        ports: ["5432:5432"]
        options: --health-cmd="pg_isready" --health-interval=10s
    env:
      DATABASE_URL: postgresql+asyncpg://postgres:test@localhost:5432/test
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.13" }
      - run: uv sync --frozen
      - run: uv run alembic upgrade head
      - run: uv run pytest tests/test_migrations.py
      - run: |
          if [ "$(uv run alembic heads | wc -l)" -gt 1 ]; then exit 1; fi

# tests/test_migrations.py
from alembic.autogenerate import compare_metadata
from alembic.migration import MigrationContext
from sqlalchemy import create_engine

def test_no_drift():
    eng = create_engine(DATABASE_URL.replace("+asyncpg", ""))
    with eng.connect() as conn:
        ctx = MigrationContext.configure(conn)
        diff = compare_metadata(ctx, Base.metadata)
        assert diff == [], f"Drift detected: {diff}"

def test_migrate_up_down():
    from alembic.config import Config
    from alembic import command
    
    cfg = Config("alembic.ini")
    cfg.set_main_option("sqlalchemy.url", DATABASE_URL)
    
    command.upgrade(cfg, "head")
    command.downgrade(cfg, "base")
    command.upgrade(cfg, "head")

if [ "$(alembic heads | wc -l)" -gt 1 ]; then
  echo "Multiple heads — merge required"
  exit 1
fi

alembic upgrade head --sql > migration.sql

import os

def test_migration_files_have_message():
    for fname in os.listdir("migrations/versions"):
        if not fname.endswith(".py"): continue
        with open(f"migrations/versions/{fname}") as f:
            content = f.read()
        # Must have docstring
        assert '"""' in content
        # Must have upgrade and downgrade
        assert "def upgrade()" in content
        assert "def downgrade()" in content

- run: alembic --name users upgrade head
- run: alembic --name analytics upgrade head
- run: pytest tests/test_drift_users.py
- run: pytest tests/test_drift_analytics.py

apiVersion: batch/v1
kind: Job
metadata:
  name: migrate-{{ .Values.image.tag }}
  annotations:
    "helm.sh/hook": pre-install,pre-upgrade
    "helm.sh/hook-weight": "0"
    "helm.sh/hook-delete-policy": before-hook-creation
spec:
  template:
    spec:
      containers:
        - name: migrate
          image: myapp:{{ .Values.image.tag }}
          command: ["alembic", "upgrade", "head"]
          envFrom:
            - secretRef: { name: app-secrets }
      restartPolicy: OnFailure
  backoffLimit: 3
  ttlSecondsAfterFinished: 300

# Health endpoint
@app.get("/migration_status")
async def status(db = Depends(get_db)):
    rev = await db.scalar(text("SELECT version_num FROM alembic_version"))
    return {"current_revision": rev, "expected": EXPECTED_HEAD}

# .pre-commit-config.yaml
- repo: local
  hooks:
    - id: alembic-single-head
      name: alembic-single-head
      entry: bash -c 'if [ "$(alembic heads | wc -l)" -gt 1 ]; then exit 1; fi'
      language: system
      pass_filenames: false

- run: alembic upgrade --sql head > migration.sql
- uses: actions/upload-artifact@v3
  with:
    name: migration-sql
    path: migration.sql

def test_no_slow_ops():
    # Read latest migration file
    latest = sorted(os.listdir("migrations/versions"))[-1]
    with open(f"migrations/versions/{latest}") as f:
        content = f.read()
    # Heuristic: warn on potentially slow ops without CONCURRENTLY
    assert "CREATE INDEX " not in content or "CONCURRENTLY" in content