FastAPI + SQLModel multi-tenant SaaS — datamodel og routing mønstre

M-Bus Gateway er bygget med FastAPI + SQLModel fra dag 1 som multi-tenant SaaS. Her er datamodel-mønstrene der giver sikker tenant-isolation.

---

SQLModel basemodel: Fælles felter

# server/src/db/base.py
from sqlmodel import SQLModel, Field
from datetime import datetime
from uuid import UUID, uuid4
from typing import Optional


class TenantBase(SQLModel):
    """
    Basismodel for alle tabeller med tenant-isolation.
    Arv fra denne — aldrig fra SQLModel direkte.
    """
    id: UUID = Field(default_factory=uuid4, primary_key=True)
    tenant_id: UUID = Field(index=True, nullable=False)

    # Audit trail (aldrig slet — rekonstruér historik):
    created_at: datetime = Field(default_factory=datetime.utcnow, nullable=False)
    updated_at: datetime = Field(
        default_factory=datetime.utcnow,
        nullable=False,
        sa_column_kwargs={"onupdate": datetime.utcnow},
    )

    # Soft delete (ALDRIG hard delete):
    deleted_at: Optional[datetime] = Field(default=None, nullable=True)
    deleted_by: Optional[UUID] = Field(default=None, nullable=True)


class Property(TenantBase, table=True):
    __tablename__ = "property"

    name: str = Field(index=True)
    address: str
    zip_code: str = Field(max_length=10)
    city: str
    country: str = Field(default="DK", max_length=2)
    heated_area_m2: float
    currency: str = Field(default="DKK", max_length=3)  # Multi-currency fra dag 1


class Unit(TenantBase, table=True):
    __tablename__ = "unit"

    property_id: UUID = Field(foreign_key="property.id", index=True)
    name: str
    floor: Optional[int] = None
    area_m2: float

---

Session factory: Async med FastAPI lifespan

# server/src/db/session.py
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlmodel import SQLModel
from server.src.core.config import settings

engine = create_async_engine(
    settings.DATABASE_URL,  # postgresql+asyncpg://...
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,     # Tjek forbindelsen før brug (håndtér idle timeouts)
    echo=settings.DEBUG,    # SQL logging i dev
)

AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,  # Undgå lazy loading efter commit i async kontekst
)


async def get_session() -> AsyncSession:
    """FastAPI dependency — ny session pr. request, lukkes automatisk."""
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise


# Lifespan: Opret tabeller ved startup (dev) / kør Alembic (prod):
from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup:
    if settings.ENV == "development":
        async with engine.begin() as conn:
            await conn.run_sync(SQLModel.metadata.create_all)
    yield
    # Shutdown:
    await engine.dispose()

---

Router mønster: Automatisk tenant-isolation

# server/src/properties/router.py
from fastapi import APIRouter, Depends, HTTPException
from sqlmodel import select
from server.src.auth.deps import get_current_user, require_role
from server.src.db.session import get_session

router = APIRouter(prefix="/properties", tags=["properties"])


@router.get("", response_model=list[PropertyOut])
async def list_properties(
    session: AsyncSession = Depends(get_session),
    user: TokenPayload = Depends(require_role("landlord", "super_admin", "external_admin")),
):
    stmt = select(Property).where(Property.deleted_at.is_(None))

    # Tenant-isolation: super_admin ser alt, andre ser kun eget:
    if user.role != "super_admin":
        if user.tenant_id is None:
            return []
        stmt = stmt.where(Property.tenant_id == user.tenant_id)

    result = await session.execute(stmt)
    return result.scalars().all()


@router.delete("/{property_id}", status_code=204)
async def delete_property(
    property_id: UUID,
    session: AsyncSession = Depends(get_session),
    user: TokenPayload = Depends(require_role("landlord", "super_admin")),
):
    property = await session.get(Property, property_id)

    # Object-level auth — returnér 404 (ikke 403) for at undgå info disclosure:
    if not property or property.deleted_at:
        raise HTTPException(404)
    if user.role != "super_admin" and property.tenant_id != user.tenant_id:
        raise HTTPException(404)

    # Soft delete — ALDRIG hard delete:
    property.deleted_at = datetime.utcnow()
    property.deleted_by = user.sub
    property.updated_at = datetime.utcnow()
    session.add(property)
    # session.commit() sker automatisk i get_session dependency

---

Many-to-many med link-tabel

# Meter installeret i lejlighed i periode:
class MeterInstallation(TenantBase, table=True):
    __tablename__ = "meter_installation"

    meter_id: UUID = Field(foreign_key="meter.id", index=True)
    unit_id: UUID = Field(foreign_key="unit.id", index=True)
    installed_at: datetime = Field(default_factory=datetime.utcnow)
    removed_at: Optional[datetime] = None
    installed_by: Optional[UUID] = None

    # Aflæsninger gemmes med meter_installation_id (IKKE meter_id direkte):
    # Sikrer at historiske aflæsninger forbliver koblet til den rigtige
    # installation-periode selv efter målerskift


class Reading(SQLModel, table=True):
    """Hypertable — ingen TenantBase (for TimescaleDB-compat)."""
    __tablename__ = "reading"

    meter_installation_id: UUID = Field(
        foreign_key="meter_installation.id",
        primary_key=True,
        index=True,
    )
    timestamp: datetime = Field(primary_key=True)  # Compound PK for hypertable
    value: float
    rssi_dbm: Optional[float] = None
    battery_level_pct: Optional[float] = None
    received_at: datetime = Field(default_factory=datetime.utcnow)

---

Test-isolering: Rollback pr. test

# server/tests/conftest.py
import pytest
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlmodel import SQLModel
from fastapi.testclient import TestClient
from server.src.main import app
from server.src.db.session import get_session

TEST_DATABASE_URL = "postgresql+asyncpg://mbus:mbus@localhost/mbus_test"

@pytest.fixture(scope="session")
def engine():
    return create_async_engine(TEST_DATABASE_URL)

@pytest.fixture(autouse=True)
async def db_session(engine):
    """
    Wrap hvert test i en transaktion der rulles tilbage.
    → Ingen test-data lækker til næste test.
    → Hurtigere end TRUNCATE (ingen disk I/O).
    """
    async with engine.connect() as conn:
        await conn.begin()
        session = AsyncSession(bind=conn)

        # Override dependency:
        async def _override():
            yield session

        app.dependency_overrides[get_session] = _override
        yield session

        # Rollback efter test:
        await session.close()
        await conn.rollback()

    app.dependency_overrides.clear()

---

Konklusion

FastAPI + SQLModel giver type-sikker, selvdokumenterende API med fuld async-support. TenantBase med tenant_id, deleted_at og audit-felter på alle tabeller er den vigtigste arkitekturbeslutning. Session-dependency med automatisk commit/rollback forenkler request-håndtering. Test-rollback-fixtures giver isolerede, hurtige tests uden database-oprydning.

Se SQLModel RBAC guide eller Alembic migrations guide.