Alembic database-migrationer med FastAPI og SQLModel

M-Bus Gateway platformen bruger Alembic til alle databaseskema-ændringer. Her er setup og mønstre der giver sikre, rollback-bare migrationer.

---

Alembic-setup med SQLModel

# server/alembic.ini (forenklet):
[alembic]
script_location = alembic
sqlalchemy.url = postgresql+psycopg2://%(DB_USER)s:%(DB_PASS)s@%(DB_HOST)s/%(DB_NAME)s

# server/alembic/env.py
from alembic import context
from sqlmodel import SQLModel
from server.src.db.models import *  # Importer ALLE modeller (metadata tracking)

config = context.config

def run_migrations_online():
    from sqlalchemy import engine_from_config
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=SQLModel.metadata,
            # Sammenlign serverens standardværdier:
            compare_server_default=True,
            # Sammenlign type-ændringer:
            compare_type=True,
        )
        with context.begin_transaction():
            context.run_migrations()

run_migrations_online()

---

Autogenerate: Find schema-ændringer automatisk

# Ændringer i models.py → Alembic finder dem automatisk:

# 1. Opret ny kolonne i models.py:
# class Meter(SQLModel, table=True):
#     mid_expiry_date: date | None = Field(default=None)  ← NY

# 2. Generér migration:
cd server
alembic revision --autogenerate -m "add_mid_expiry_date_to_meter"

# Output:
# Generating /server/alembic/versions/0042_add_mid_expiry_date_to_meter.py

# 3. Review genereret migration (ALTID!):
cat alembic/versions/0042_add_mid_expiry_date_to_meter.py

# Genereret migration (review kritisk — autogenerate er ikke perfekt):
# server/alembic/versions/0042_add_mid_expiry_date_to_meter.py

def upgrade() -> None:
    op.add_column(
        "meter",
        sa.Column("mid_expiry_date", sa.Date(), nullable=True),
    )

def downgrade() -> None:
    op.drop_column("meter", "mid_expiry_date")

---

Data-migration: Udfyld nye kolonner

# Når en ny NOT NULL-kolonne kræver en default-værdi for eksisterende rækker:
# server/alembic/versions/0043_populate_mid_expiry_date.py

from alembic import op
import sqlalchemy as sa
from datetime import date, timedelta

def upgrade() -> None:
    # Tilføj kolonne som nullable FØRST:
    op.add_column(
        "meter",
        sa.Column("mid_expiry_date", sa.Date(), nullable=True),
    )

    # Udfyld eksisterende rækker (data-migration):
    connection = op.get_bind()
    connection.execute(
        sa.text("""
            UPDATE meter
            SET mid_expiry_date = installation_date + INTERVAL '10 years'
            WHERE mid_expiry_date IS NULL
              AND meter_type = 'hca'
        """)
    )
    connection.execute(
        sa.text("""
            UPDATE meter
            SET mid_expiry_date = installation_date + INTERVAL '5 years'
            WHERE mid_expiry_date IS NULL
              AND meter_type IN ('heat', 'water')
        """)
    )

    # Gør kolonne NOT NULL nu data er udfyldt:
    op.alter_column("meter", "mid_expiry_date", nullable=False)


def downgrade() -> None:
    op.drop_column("meter", "mid_expiry_date")

---

Multi-tenant-safe migrations

# KRITISK: Alle nye tabeller SKAL have tenant_id fra dag 1.
# Alembic hjælper ikke med dette — det er en manuel tjek.

def upgrade() -> None:
    op.create_table(
        "new_table",
        sa.Column("id", sa.UUID(), nullable=False, primary_key=True),
        # tenant_id er OBLIGATORISK — aldrig glem det:
        sa.Column("tenant_id", sa.UUID(), nullable=False),
        sa.Column("created_at", sa.DateTime(), nullable=False,
                  server_default=sa.text("NOW()")),
        sa.Column("updated_at", sa.DateTime(), nullable=False,
                  server_default=sa.text("NOW()")),
        sa.Column("deleted_at", sa.DateTime(), nullable=True),
        sa.Column("deleted_by", sa.UUID(), nullable=True),
        sa.ForeignKeyConstraint(["tenant_id"], ["tenant.id"]),
    )

    # Index på tenant_id er kritisk for performance:
    op.create_index("ix_new_table_tenant_id", "new_table", ["tenant_id"])

---

Zero-downtime migrations

# Problematisk: Tilføj NOT NULL-kolonne uden default
# → PostgreSQL låser hele tabellen under ALTER TABLE

# UNDGÅ:
def upgrade_bad() -> None:
    op.add_column("reading", sa.Column("source", sa.String(), nullable=False))
    # ↑ Kaster fejl hvis tabellen har rækker (ingen default)
    # ↑ Låser tabellen (nedetid)

# KORREKT (3-trin, zero-downtime):
# Migration 1: Tilføj nullable:
def upgrade_step1() -> None:
    op.add_column("reading", sa.Column("source", sa.String(), nullable=True))

# Deployment: App bruger source-felt (skriver nullable eller "mqtt")

# Migration 2: Backfill:
def upgrade_step2() -> None:
    op.execute("UPDATE reading SET source = 'mqtt' WHERE source IS NULL")

# Migration 3: Gør NOT NULL:
def upgrade_step3() -> None:
    op.alter_column("reading", "source", nullable=False)

---

CI/CD integration

# .github/workflows/ci.yml — kør migrationer inden tests:

- name: Run migrations
  env:
    DATABASE_URL: postgresql+asyncpg://test:test@localhost:5432/mbus_test
  run: |
    cd server
    # Brug psycopg2 URL til Alembic (ikke asyncpg):
    ALEMBIC_DB_URL="postgresql+psycopg2://test:test@localhost:5432/mbus_test" \
      alembic upgrade head

- name: Verify no pending migrations
  run: |
    cd server
    # Fejl hvis der er uapplicerede autogenerate-ændringer:
    alembic check
    # Returnerer exit code 1 hvis models.py er ude af sync med DB

---

Rollback-strategi

# Rollback én migration:
alembic downgrade -1

# Rollback til specifik revision:
alembic downgrade 0041

# Se nuværende revision:
alembic current

# Se migration-historik:
alembic history --verbose

# I produktion — rollback med forsigtighed:
# 1. Stop server (undgå concurrent skrivning under rollback)
# 2. alembic downgrade -1
# 3. Deploy forrige kodeversion
# 4. Start server igen

---

Konklusion

Alembic med --autogenerate finder skema-ændringer automatisk, men migrationer skal altid reviewes inden apply. Data-migrationer kræver 3-trins tilgang for zero-downtime. Alle nye tabeller skal have tenant_id, created_at, updated_at, deleted_at — Alembic hverken tjekker eller kræver dette, det er en arkitektur-regel der håndhæves ved code review.

Se FastAPI SQLModel guide eller asyncpg guide.