M-Bus Gateway
← Tilbage til blog
· python· pydantic· settings· konfiguration· SecretStr· environment· Docker· IoT· SaaS

Python Pydantic Settings — konfiguration til IoT og SaaS

Byg type-safe konfiguration med Pydantic BaseSettings: SecretStr, lru_cache singleton, nested settings, Docker Secrets, environment variable validation og test-overrides.

Af M-Bus Gateway

Pydantic Settings giver type-valideret konfiguration fra environment variables, .env-filer og Docker Secrets — uden hardkodede værdier. Her er mønstrene der bruges i produktion.


Basal BaseSettings

# server/src/config.py
from pydantic import SecretStr, PostgresDsn, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,        # DATABASE_URL = database_url
        extra="ignore",              # Ignorer ukendte env-vars
    )

    # Database
    database_url: PostgresDsn

    # JWT
    jwt_secret: SecretStr            # Aldrig eksponeret i logs eller repr()
    jwt_algorithm: str = "HS256"
    jwt_expire_minutes: int = 60

    # Brevo email
    brevo_api_key: SecretStr

    # Stripe
    stripe_secret_key: SecretStr
    stripe_webhook_secret: SecretStr

    # Platform
    platform_name: str = "M-Bus Gateway"
    environment: str = "development"
    debug: bool = False

    @field_validator("environment")
    @classmethod
    def validate_environment(cls, v: str) -> str:
        allowed = {"development", "staging", "production"}
        if v not in allowed:
            raise ValueError(f"environment skal være én af: {allowed}")
        return v

    @property
    def is_production(self) -> bool:
        return self.environment == "production"

lru_cache singleton

# server/src/config.py (fortsat)
from functools import lru_cache


@lru_cache(maxsize=1)
def get_settings() -> Settings:
    """
    Singleton: Læses kun fra disk én gang.
    lru_cache sikrer at Settings() kun instantieres én gang,
    selv ved mange parallelle requests.
    """
    return Settings()


# Brug i FastAPI dependency injection:
from fastapi import Depends
from typing import Annotated

def get_db_url(settings: Annotated[Settings, Depends(get_settings)]) -> str:
    return str(settings.database_url)

# Direkte import:
settings = get_settings()

SecretStr — secrets uden lækage

# SecretStr forhindrer accidentel eksponering i logs og fejlbeskeder

from pydantic import SecretStr

secret = SecretStr("super-hemmelig-nøgle")

# Disse viser ALDRIG den faktiske værdi:
print(secret)            # **********
print(repr(secret))      # SecretStr('**********')
str(secret)              # **********

# Kun .get_secret_value() afslører værdien:
actual_value = secret.get_secret_value()    # "super-hemmelig-nøgle"

# Brug i httpx-klient:
headers = {
    "Authorization": f"Bearer {settings.stripe_secret_key.get_secret_value()}"
}

# Brug i JWT:
import jwt
token = jwt.encode(
    payload,
    settings.jwt_secret.get_secret_value(),
    algorithm=settings.jwt_algorithm,
)

Nested settings (subsektioner)

# server/src/config.py — kompleks konfiguration med subsektioner

from pydantic import BaseModel
from pydantic_settings import BaseSettings, SettingsConfigDict


class MQTTSettings(BaseModel):
    host: str = "localhost"
    port: int = 1883
    tls_port: int = 8883
    client_id: str = "mbus-server"
    use_tls: bool = False


class StorageSettings(BaseModel):
    endpoint_url: str = "https://fsn1.your-objectstorage.com"
    bucket_name: str = "mbus-pdfs"
    access_key: str = ""
    secret_key: str = ""


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_nested_delimiter="__",    # MQTT__HOST=... → mqtt.host
    )

    database_url: str = "postgresql+asyncpg://localhost/mbus"
    mqtt: MQTTSettings = MQTTSettings()
    storage: StorageSettings = StorageSettings()

# .env-fil:
# MQTT__HOST=178.105.90.8
# MQTT__PORT=8883
# MQTT__USE_TLS=true
# STORAGE__BUCKET_NAME=mbus-pdfs-prod

Docker Secrets integration

# server/src/config.py — Docker Secrets via fil-baseret hemmeligheder

from pathlib import Path
from pydantic import SecretStr, model_validator
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    # Understøtter både environment variable og Docker Secret fil:
    # DATABASE_URL=postgresql://... (env var)
    # ELLER: /run/secrets/database_url (Docker Secret)

    database_url: str = ""
    jwt_secret: SecretStr = SecretStr("")

    @model_validator(mode="after")
    def load_docker_secrets(self) -> "Settings":
        """
        Docker Swarm gemmer secrets som filer under /run/secrets/.
        Læs fil hvis environment variable er tom.
        """
        secrets_dir = Path("/run/secrets")

        if not self.database_url:
            secret_file = secrets_dir / "database_url"
            if secret_file.exists():
                self.database_url = secret_file.read_text().strip()

        if not self.jwt_secret.get_secret_value():
            secret_file = secrets_dir / "jwt_secret"
            if secret_file.exists():
                self.jwt_secret = SecretStr(secret_file.read_text().strip())

        return self

# docker-compose.yml:
# secrets:
#   database_url:
#     file: ./secrets/database_url.txt
#   jwt_secret:
#     file: ./secrets/jwt_secret.txt
# services:
#   server:
#     secrets:
#       - database_url
#       - jwt_secret

Gateway-konfiguration (SQLite-baseret)

# gateway/src/config.py — Pi-konfiguration fra SQLite (ikke .env)
# Gateway gemmer konfiguration i lokal SQLite for offline-adgang

import aiosqlite
from dataclasses import dataclass
from typing import Optional


@dataclass
class GatewayConfig:
    gateway_id: str = "GW-UNKNOWN"
    mqtt_host: str = "178.105.90.8"
    mqtt_port: int = 8883
    mqtt_use_tls: bool = True
    send_cron: str = "0 6 * * *"
    wmbus_mode: str = "c1"
    log_level: str = "INFO"
    cloudflare_token: str = ""
    platform_api_url: str = ""


async def load_gateway_config(db_path: str = "/var/lib/mbus-gateway/config.db") -> GatewayConfig:
    """Læs konfiguration fra SQLite config-tabel."""
    config = GatewayConfig()

    async with aiosqlite.connect(db_path) as db:
        async with db.execute("SELECT key, value FROM config") as cursor:
            async for row in cursor:
                key, value = row
                if hasattr(config, key.lower()):
                    field_type = type(getattr(config, key.lower()))
                    if field_type == bool:
                        setattr(config, key.lower(), value.lower() == "true")
                    elif field_type == int:
                        setattr(config, key.lower(), int(value))
                    else:
                        setattr(config, key.lower(), value)

    return config

Test-overrides

# server/tests/conftest.py — override settings i tests

import pytest
from unittest.mock import patch
from server.src.config import Settings, get_settings


@pytest.fixture(autouse=True)
def override_settings(monkeypatch):
    """
    Override settings for alle tests.
    monkeypatch.setenv sætter environment variables FØR Settings() instantieres.
    """
    monkeypatch.setenv("DATABASE_URL", "postgresql+asyncpg://localhost/test_mbus")
    monkeypatch.setenv("JWT_SECRET", "test-secret-key-32-chars-minimum")
    monkeypatch.setenv("BREVO_API_KEY", "test-brevo-key")
    monkeypatch.setenv("STRIPE_SECRET_KEY", "sk_test_fake")
    monkeypatch.setenv("STRIPE_WEBHOOK_SECRET", "whsec_test_fake")
    monkeypatch.setenv("ENVIRONMENT", "development")

    # Ryd lru_cache så test-settings bruges:
    get_settings.cache_clear()
    yield
    get_settings.cache_clear()


@pytest.fixture
def test_settings() -> Settings:
    return get_settings()


# Alternativ: Direkte dependency override i FastAPI test-client:
from fastapi.testclient import TestClient
from server.src.main import app
from server.src.config import get_settings


def get_test_settings() -> Settings:
    return Settings(
        database_url="postgresql+asyncpg://localhost/test",
        jwt_secret="test-secret",
        brevo_api_key="test",
        stripe_secret_key="sk_test",
        stripe_webhook_secret="whsec_test",
    )


app.dependency_overrides[get_settings] = get_test_settings
client = TestClient(app)

Validering og fejlhåndtering ved opstart

# server/src/main.py — fail-fast ved manglende konfiguration

from contextlib import asynccontextmanager
from fastapi import FastAPI
from pydantic import ValidationError
from server.src.config import get_settings
import structlog

log = structlog.get_logger()


@asynccontextmanager
async def lifespan(app: FastAPI):
    try:
        settings = get_settings()
        log.info("configuration_loaded", environment=settings.environment)

        if settings.is_production and settings.debug:
            raise ValueError("debug=True er ikke tilladt i production")

    except ValidationError as e:
        log.error("configuration_error", errors=e.errors())
        raise SystemExit(1) from e    # Fail-fast: start ikke med ugyldig config

    yield    # Application kører

    log.info("shutting_down")

Konklusion

Pydantic BaseSettings + SecretStr + lru_cache er standardmønsteret til type-sikker konfiguration der aldrig eksponerer secrets i logs. env_nested_delimiter="__" giver rene subsektioner. Docker Secrets understøttes via model_validator. Test-overrides via monkeypatch.setenv + cache_clear() sikrer isolation mellem tests.

Se Python httpx guide eller Python dataclasses guide.