Python functools — praktisk guide til IoT og SaaS

functools er et af Pythons mest undervurderede standardbiblioteker. Her er mønstrene der bruges dagligt i IoT-gateways og SaaS-backends.

---

lru_cache — memoization til dyre opslag

# server/src/config.py
from functools import lru_cache
from pydantic_settings import BaseSettings


@lru_cache(maxsize=1)
def get_settings() -> BaseSettings:
    """
    Settings indlæses én gang — derefter fra cache.
    maxsize=1: Kun ét unikt resultat gemmes (ingen args).
    """
    return BaseSettings()


# Ryd cache i tests:
# get_settings.cache_clear()

# Cache-info (til debugging):
# get_settings.cache_info()
# → CacheInfo(hits=42, misses=1, maxsize=1, currsize=1)


@lru_cache(maxsize=256)
def get_zip_code_municipality(zip_code: str) -> str | None:
    """
    ZIP → kommunekode opslag. Resultater caches pr. ZIP.
    maxsize=256: Passer til antal postnumre i DK (574 i alt).
    """
    from server.src.data.zip_codes import ZIP_TO_MUNICIPALITY
    return ZIP_TO_MUNICIPALITY.get(zip_code)


# Vigtig regel: lru_cache virker kun på hashable argumenter.
# list, dict, set → TypeError. Brug tuple eller frozenset i stedet.

---

partial — route og task factories

# server/src/api/router_factory.py
from functools import partial
from fastapi import Depends
from server.src.auth.deps import require_role


# partial lader dig "forudfylde" argumenter til en funktion:

require_landlord = partial(require_role, allowed_roles=["landlord", "super_admin"])
require_admin = partial(require_role, allowed_roles=["super_admin"])
require_tech = partial(require_role, allowed_roles=["technician", "super_admin"])

# Brug i router:
# @router.get("/properties", dependencies=[Depends(require_landlord)])


# partial til Celery-tasks med forudfyldt parameter:
from celery import Celery

app = Celery()

def send_email_task(email_type: str, recipient: str, **kwargs) -> None:
    pass    # Email-logik

# Factory: Forudfyld email_type
send_settlement_email = partial(send_email_task, email_type="settlement")
send_alarm_email = partial(send_email_task, email_type="alarm")
send_reminder_email = partial(send_email_task, email_type="reminder")

# Brug:
# send_settlement_email(recipient="lejer@example.com", settlement_id="S-001")

---

reduce — akkumulering af fordelingsresultater

# server/src/distribution/aggregator.py
from functools import reduce
from decimal import Decimal


def total_allocated(allocations: dict[str, Decimal]) -> Decimal:
    """Summér alle fordelingsbeløb via reduce."""
    return reduce(lambda acc, v: acc + v, allocations.values(), Decimal("0"))


def combine_distributions(
    distributions: list[dict[str, Decimal]]
) -> dict[str, Decimal]:
    """
    Kombinér flere fordelingsresultater (f.eks. varme + vand + el).
    reduce folder listen til ét samlet dict med summerede beløb.
    """
    def merge(acc: dict[str, Decimal], dist: dict[str, Decimal]) -> dict[str, Decimal]:
        return {
            unit_id: acc.get(unit_id, Decimal("0")) + amount
            for unit_id, amount in dist.items()
        }

    return reduce(merge, distributions, {})


# Eksempel:
heat_dist = {"U-01": Decimal("1200.00"), "U-02": Decimal("980.00")}
water_dist = {"U-01": Decimal("340.00"), "U-02": Decimal("290.00")}
combined = combine_distributions([heat_dist, water_dist])
# → {"U-01": Decimal("1540.00"), "U-02": Decimal("1270.00")}

---

wraps — korrekte dekoratorer

# server/src/api/decorators.py
from functools import wraps
from typing import Callable, Any
import time
import structlog

log = structlog.get_logger()


def log_duration(func: Callable) -> Callable:
    """
    Dekorator der logger kaldetid.
    @wraps bevarer __name__, __doc__ og __wrapped__ på den originale funktion.
    Uden @wraps: FastAPI/OpenAPI vil se "wrapper" i stedet for "get_properties".
    """
    @wraps(func)
    async def wrapper(*args: Any, **kwargs: Any) -> Any:
        start = time.monotonic()
        try:
            result = await func(*args, **kwargs)
            return result
        finally:
            duration_ms = (time.monotonic() - start) * 1000
            log.info(
                "endpoint_duration",
                function=func.__name__,
                duration_ms=round(duration_ms, 1),
            )

    return wrapper


def require_feature_flag(flag: str) -> Callable:
    """
    Fabrik der returnerer en dekorator med forudfyldt flag-navn.
    Kombinerer partial-mønsteret med @wraps.
    """
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        async def wrapper(*args: Any, **kwargs: Any) -> Any:
            from server.src.config import get_settings
            settings = get_settings()
            if not getattr(settings, flag, False):
                from fastapi import HTTPException
                raise HTTPException(status_code=404, detail="Feature not enabled")
            return await func(*args, **kwargs)
        return wrapper
    return decorator

# Brug:
# @require_feature_flag("oioubl_enabled")
# async def get_oioubl_invoice(settlement_id: str): ...

---

cached_property — lazy instans-attributter

# gateway/src/wmbus/meter.py
from functools import cached_property
from dataclasses import dataclass


@dataclass
class MeterReading:
    """
    cached_property: Beregnes første gang den tilgås,
    derefter gemt som instansattribut (ingen genberegning).
    Virker kun på klasser der tillader __dict__ (ikke frozen dataclasses).
    """
    meter_id: str
    total_kwh_raw: float
    flow_temp_c: float
    return_temp_c: float

    @cached_property
    def delta_temp(self) -> float:
        """ΔT = flow - retur. Beregnes kun én gang."""
        return self.flow_temp_c - self.return_temp_c

    @cached_property
    def is_suspicious(self) -> bool:
        """
        Simpel anomali-check der kombinerer multiple kriterier.
        cached_property er ideel her: dyrt at genberegne, bruges sjældent.
        """
        return (
            self.delta_temp < 5.0           # For lille ΔT → hydraulisk problem
            or self.total_kwh_raw < 0       # Negativt forbrug → backwards alarm
            or self.total_kwh_raw > 50000   # Urealistisk højt → spike alarm
        )

---

total_ordering — sammenligning af målere

# server/src/models/meter.py
from functools import total_ordering
from datetime import date


@total_ordering
class MeterMidStatus:
    """
    total_ordering: Definer __eq__ og ét sammenligningsoperator,
    resten genereres automatisk (__le__, __gt__, __ge__).
    Bruges til at sortere målere efter MID-udløbsdato.
    """
    def __init__(self, meter_id: str, mid_expiry: date):
        self.meter_id = meter_id
        self.mid_expiry = mid_expiry

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, MeterMidStatus):
            return NotImplemented
        return self.mid_expiry == other.mid_expiry

    def __lt__(self, other: "MeterMidStatus") -> bool:
        if not isinstance(other, MeterMidStatus):
            return NotImplemented
        return self.mid_expiry < other.mid_expiry


# Sortér liste af målere efter MID-udløb (tidligst udløb først):
meters = [
    MeterMidStatus("KAM-001", date(2028, 6, 1)),
    MeterMidStatus("ELS-002", date(2026, 3, 15)),
    MeterMidStatus("TCH-003", date(2027, 9, 1)),
]
sorted_meters = sorted(meters)    # ELS-002, TCH-003, KAM-001

---

Konklusion

functools giver seks kraftfulde værktøjer: lru_cache til singleton og memoization, partial til route-factories og pre-filled Celery tasks, reduce til distribution-aggregering, wraps til korrekte dekoratorer (kritisk for FastAPI/OpenAPI), cached_property til lazy beregnede instansattributter og total_ordering til automatisk genererede sammenligningsoperatorer. Alle seks bruges i produktionskoden.

Se Python lru_cache guide eller Python dataclasses guide.