Alembic zero-downtime database migrationer — FastAPI og PostgreSQL

Database migrationer er den farligste del af et SaaS-deploy — forkert udført kan de låse tabeller og forårsage nedetid. Her er zero-downtime mønstrene.

---

Expand/Contract mønster

Zero-downtime kræver at migrationer er bakwards-kompatible:

EXPAND (tilføj):
  → Tilføj ny kolonne (nullable eller med default)
  → Tilføj ny tabel
  → Tilføj ny index (CONCURRENTLY)
  → Deploy ny kode der læser fra BEGGE steder

CONTRACT (fjern gammel):
  → Fjern gammel kolonne/tabel (når ingen kode bruger den)
  → Deploy som separat migration efter næste deploy

Eksempel: Omdøb kolonne (3-fase):
  Fase 1: Tilføj ny kolonne, kopi data, opdatér kode til at skrive begge
  Fase 2: Deploy — kode skriver til begge, læser fra ny
  Fase 3: Fjern gammel kolonne i separat migration

---

Alembic konfiguration med asyncpg

# server/alembic/env.py

import asyncio
from logging.config import fileConfig
from alembic import context
from sqlalchemy.ext.asyncio import create_async_engine
from sqlmodel import SQLModel

# Import alle models for at registrere metadata
import server.src.db.models  # noqa: F401

target_metadata = SQLModel.metadata

def run_migrations_online() -> None:
    connectable = create_async_engine(
        context.config.get_main_option("sqlalchemy.url"),
        echo=False,
    )

    async def do_run():
        async with connectable.connect() as connection:
            await connection.run_sync(_run_migrations)

    asyncio.get_event_loop().run_until_complete(do_run())

def _run_migrations(connection):
    context.configure(
        connection=connection,
        target_metadata=target_metadata,
        compare_type=True,
        compare_server_default=True,
        include_schemas=True,
        # Transaction per migration file:
        transaction_per_migration=True,
    )
    with context.begin_transaction():
        context.run_migrations()

---

Ikke-destruktive ændringer

# Fase 1: Tilføj ny kolonne (nullable — zero-downtime)
# server/alembic/versions/0042_add_preferred_language.py

def upgrade() -> None:
    op.add_column(
        "user",
        sa.Column(
            "preferred_language",
            sa.String(5),
            nullable=True,  # IKKE NOT NULL uden default
            server_default=None,
        ),
    )
    # Backfill eksisterende rækker asynkront (Celery task, ikke migration)

def downgrade() -> None:
    op.drop_column("user", "preferred_language")

# Fase 2 (separat migration, næste deploy):
# Sæt NOT NULL + default når alle rækker er backfilled
def upgrade() -> None:
    op.alter_column(
        "user",
        "preferred_language",
        nullable=False,
        server_default="da",
    )

---

Index CONCURRENTLY (undgå table lock)

# Opret index uden at låse tabellen:
def upgrade() -> None:
    # Standard: CREATE INDEX — låser tabellen (bloker writes)!
    # Zero-downtime: CREATE INDEX CONCURRENTLY
    op.execute(
        "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_reading_meter_installation_id "
        "ON reading (meter_installation_id)"
    )
    # NB: CONCURRENTLY kræver at migrationen IKKE kører i transaktion:
    # Brug transaction_per_migration=False for denne migration

def downgrade() -> None:
    op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_reading_meter_installation_id")

# alembic/env.py override pr. revision:
def _run_migrations(connection):
    context.configure(
        connection=connection,
        target_metadata=target_metadata,
        # Deaktivér transaction for CONCURRENTLY index:
        transaction_per_migration=True,
    )

---

Batch operations til SQLite (tests)

# SQLite understøtter ikke ALTER COLUMN/DROP COLUMN direkte
# Alembic batch mode håndterer dette:

def upgrade() -> None:
    with op.batch_alter_table("meter") as batch_op:
        batch_op.add_column(
            sa.Column("last_seen_at", sa.DateTime(timezone=True), nullable=True)
        )
        batch_op.alter_column(
            "status",
            new_column_name="meter_status",  # Omdøb
        )
        batch_op.drop_column("legacy_field")

---

Data migration (backfill)

# server/alembic/versions/0043_backfill_currency.py

from alembic import op
import sqlalchemy as sa

def upgrade() -> None:
    # Tilføj kolonne
    op.add_column(
        "settlement_line",
        sa.Column("currency", sa.String(3), nullable=True),
    )

    # Backfill via SQL (hurtigere end ORM)
    op.execute("""
        UPDATE settlement_line sl
        SET currency = p.currency
        FROM settlement s
        JOIN annual_input ai ON ai.id = s.annual_input_id
        JOIN property p ON p.id = ai.property_id
        WHERE sl.settlement_id = s.id
        AND sl.currency IS NULL
    """)

    # Sæt NOT NULL EFTER backfill
    op.alter_column("settlement_line", "currency", nullable=False)

def downgrade() -> None:
    op.drop_column("settlement_line", "currency")

---

CI integration

# .github/workflows/test-server.yml

- name: Run Alembic migrations
  env:
    DATABASE_URL: postgresql+asyncpg://postgres:postgres@localhost:5432/testdb
  run: |
    alembic upgrade head

- name: Verify no pending migrations
  run: |
    # Fejler hvis autogenerate finder usynkroniserede modeller:
    alembic check
    # Alternativt:
    python -c "
    from alembic.config import Config
    from alembic.script import ScriptDirectory
    from alembic.runtime.environment import EnvironmentContext
    # Verificer at head migration matcher SQLModel metadata
    "

- name: Run tests
  run: pytest server/tests/ -x -q

---

Rollback-strategi

# Altid skriv downgrade() — selv for simple tilføjelser:

def upgrade() -> None:
    op.add_column("gateway", sa.Column("config", sa.JSON, nullable=True))
    op.create_index("ix_gateway_tenant_id", "gateway", ["tenant_id"])

def downgrade() -> None:
    op.drop_index("ix_gateway_tenant_id")
    op.drop_column("gateway", "config")

# Emergency rollback procedure:
# 1. alembic downgrade -1    (én revision tilbage)
# 2. alembic downgrade <revision_id>  (til specifik revision)
# 3. alembic current         (vis nuværende revision)
# 4. alembic history         (vis alle revisioner)

---

Naming convention

# server/alembic/env.py

NAMING_CONVENTION = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s",
}

metadata = SQLModel.metadata
metadata.naming_convention = NAMING_CONVENTION
# Gør autogenerate's ALTER TABLE navne forudsigelige og konfliktsikre

---

Konklusion

Zero-downtime Alembic migrationer kræver expand/contract mønstret: tilføj nullable kolonner og CONCURRENTLY indexes i én deploy, fjern forældede kolonner i næste. alembic check i CI verificerer at SQLModel metadata og database er synkroniserede. Altid skriv downgrade() for emergency rollback. transaction_per_migration=True giver atomisk per-fil rollback ved fejl.

Se FastAPI multi-tenant guide eller GitHub Actions CI guide.