Pydantic datavalidering i FastAPI — mønstre for IoT og SaaS

Pydantic v2 er fundamentet for FastAPI's automatiske validering og dokumentation. Her er mønstrene der giver robust API-design til IoT-platforme med heterogene data.

---

Request vs. Response models: Adskil dem altid

# server/src/properties/schemas.py
from pydantic import BaseModel, Field, field_validator
from typing import Optional
from uuid import UUID
from datetime import datetime

# CREATE (hvad klienten sender):
class PropertyCreate(BaseModel):
    name: str = Field(min_length=1, max_length=200)
    address: str = Field(min_length=5, max_length=500)
    bbr_id: Optional[str] = Field(default=None, pattern=r"^\d{15}$")
    area_m2: float = Field(gt=0, le=100_000)
    unit_count: int = Field(gt=0, le=10_000)
    # tenant_id sættes aldrig af klient — kun fra JWT

# UPDATE (partial — alle felter valgfrie):
class PropertyUpdate(BaseModel):
    name: Optional[str] = Field(default=None, min_length=1, max_length=200)
    address: Optional[str] = Field(default=None, min_length=5, max_length=500)
    area_m2: Optional[float] = Field(default=None, gt=0)
    # BBR-ID kan ikke ændres efter oprettelse

# RESPONSE (hvad API'et returnerer):
class PropertyOut(BaseModel):
    id: UUID
    name: str
    address: str
    bbr_id: Optional[str]
    area_m2: float
    unit_count: int
    tenant_id: UUID
    created_at: datetime
    updated_at: datetime
    # deleted_at eksponeres ALDRIG i API

    model_config = {"from_attributes": True}  # SQLModel/ORM compat

---

Custom validators: BBR-ID og CVR

from pydantic import field_validator, model_validator

class PropertyCreate(BaseModel):
    name: str
    bbr_id: Optional[str] = None
    cvr_number: Optional[str] = None
    area_m2: float

    @field_validator("bbr_id")
    @classmethod
    def validate_bbr_id(cls, v: Optional[str]) -> Optional[str]:
        if v is None:
            return v
        if not v.isdigit() or len(v) != 15:
            raise ValueError("BBR-ID skal være 15 cifre")
        return v

    @field_validator("cvr_number")
    @classmethod
    def validate_cvr(cls, v: Optional[str]) -> Optional[str]:
        if v is None:
            return v
        # CVR: 8 cifre
        cleaned = v.replace("-", "").replace(" ", "")
        if not cleaned.isdigit() or len(cleaned) != 8:
            raise ValueError("CVR-nummer skal være 8 cifre")
        return cleaned

    @model_validator(mode="after")
    def validate_area_consistency(self) -> "PropertyCreate":
        # Forretningsregel: Areal skal passe til enhedsantal
        if self.area_m2 < self.unit_count * 20:
            raise ValueError(
                f"Areal {self.area_m2}m² virker lavt for {self.unit_count} enheder"
                f" (min. 20 m²/enhed)"
            )
        return self

---

Discriminated unions: IoT payload-parsing

wM-Bus data kan komme i mange format afhænging af målertype:

from pydantic import BaseModel, Discriminator, Tag
from typing import Annotated, Union, Literal

class HCAReading(BaseModel):
    meter_type: Literal["hca"]
    current_hca: int
    previous_hca: int
    set_date: str
    status: str

class HeatMeterReading(BaseModel):
    meter_type: Literal["heat"]
    total_energy_kwh: float
    current_power_kw: Optional[float]
    flow_temperature_c: Optional[float]
    return_temperature_c: Optional[float]
    status: str

class WaterMeterReading(BaseModel):
    meter_type: Literal["water"]
    total_water_m3: float
    alarm_bits: int = 0
    status: str

# Discriminated union — Pydantic vælger korrekt model baseret på meter_type:
AnyReading = Annotated[
    Union[
        Annotated[HCAReading, Tag("hca")],
        Annotated[HeatMeterReading, Tag("heat")],
        Annotated[WaterMeterReading, Tag("water")],
    ],
    Discriminator("meter_type"),
]

class MQTTPayload(BaseModel):
    gateway_id: str
    timestamp: datetime
    readings: list[AnyReading]

# Brug:
payload = MQTTPayload.model_validate(raw_json)
for reading in payload.readings:
    if isinstance(reading, HCAReading):
        process_hca(reading.current_hca)
    elif isinstance(reading, WaterMeterReading):
        check_leak(reading.alarm_bits)

---

Strict mode: Ingen stille type-konvertering

from pydantic import BaseModel, ConfigDict

class ReadingCreate(BaseModel):
    model_config = ConfigDict(strict=True)  # Ingen implicit casting

    value: float      # "342" → ValueError (i strict mode)
    timestamp: datetime  # Kun ISO 8601 strenge — ikke timestamps

# I lax mode (default):
ReadingLax(value="342", timestamp=1716508800)  # Virker
# I strict mode:
ReadingStrict(value="342", timestamp=1716508800)  # ValueError

Strict mode anbefales for interne API-kald. Lax mode er OK for bruger-input.

---

Response model excludes: Skjul interne felter

from fastapi import APIRouter
from pydantic import BaseModel

router = APIRouter()

class UserOut(BaseModel):
    id: UUID
    email: str
    full_name: str
    role: str
    created_at: datetime
    # Aldrig eksponér: password_hash, aes_key_encrypted, deleted_at

@router.get("/me", response_model=UserOut)
async def get_me(user: User = Depends(get_current_user)):
    # FastAPI filtrerer automatisk til UserOut-felter:
    return user  # Returnerer User SQLModel — kun UserOut-felter eksponeres

---

Pagination schema

from pydantic import BaseModel, Field
from typing import Generic, TypeVar

T = TypeVar("T")

class Page(BaseModel, Generic[T]):
    data: list[T]
    total: int
    page: int
    per_page: int
    has_more: bool
    next_cursor: Optional[str] = None

class PageParams(BaseModel):
    page: int = Field(default=1, ge=1)
    per_page: int = Field(default=50, ge=1, le=1000)

# Brug:
@router.get("/properties", response_model=Page[PropertyOut])
async def list_properties(
    params: PageParams = Depends(),
    user: User = Depends(get_current_user),
) -> Page[PropertyOut]:
    ...

---

Konklusion

Pydantic v2 i FastAPI giver automatisk validering, OpenAPI-dokumentation og type safety i ét. Nøglemønstre for IoT-platforme: Separate Create/Update/Out-schemas, custom validators for domænespecifikke regler, discriminated unions for heterogene IoT-payloads og response_model for automatisk datamaskering. Strict mode anbefales for interne API-kald.

Se FastAPI multi-tenant guide eller API-katalog.