· Python· typing· type hints· TypeAlias· TypeGuard· Self· TypeVarTuple· FastAPI· IoT
Python typing_extensions — avancerede type hints til IoT platform
Python typing_extensions guide: TypeAlias, TypeGuard, Self, Required/NotRequired, Unpack, TypeVarTuple og overload til IoT SaaS platform med FastAPI og Pydantic.
Af M-Bus Gateway
typing_extensions backporterer Python 3.12+ type-features til Python 3.11. Her er de mest nyttige til IoT-platformen.
TypeAlias — navngivne type-aliasser
# server/src/types.py
from typing import TYPE_CHECKING
from typing_extensions import TypeAlias
import uuid
from decimal import Decimal
# Klare type-aliasser frem for primitive typer overalt:
TenantId: TypeAlias = uuid.UUID
PropertyId: TypeAlias = uuid.UUID
UnitId: TypeAlias = uuid.UUID
MeterInstallationId: TypeAlias = uuid.UUID
GatewayId: TypeAlias = str # "GW-0001" format
AmountDKK: TypeAlias = Decimal
AmountOere: TypeAlias = int # Stripe øre-format
KwhValue: TypeAlias = Decimal
HcaUnits: TypeAlias = Decimal
# Brug:
def calculate_share(
unit_id: UnitId,
total_cost: AmountDKK,
hca_units: HcaUnits,
) -> AmountDKK:
...
# Langt klarere end:
def calculate_share(
unit_id: uuid.UUID,
total_cost: Decimal,
hca_units: Decimal,
) -> Decimal: ...
TypeGuard — runtime type-verifikation
# gateway/src/wmbus/validator.py
from typing_extensions import TypeGuard
from typing import Any
class ValidTelegram(dict):
"""Markørklasse for validerede wM-Bus telegrammer."""
pass
def is_valid_telegram(data: Any) -> TypeGuard[dict]:
"""
Verificér at JSON-data er et gyldigt wM-Bus telegram.
TypeGuard fortæller type-checker at data er dict efter tjek.
"""
if not isinstance(data, dict):
return False
required_fields = {"id", "media", "meter", "timestamp", "value"}
return required_fields.issubset(data.keys())
def is_heat_telegram(data: dict) -> TypeGuard[dict]:
"""Verificér at telegram er fra varmemåler."""
return (
isinstance(data, dict)
and data.get("media") in {"heat", "heat_inlet", "warm_water"}
and "value_mwh" in data
)
# Brug — type checker ved at data er dict inden for if-blokken:
async def process_telegram(raw: Any) -> None:
if not is_valid_telegram(raw):
log.warning("telegram.invalid", data=str(raw)[:100])
return
# Her ved type-checker: raw er dict
meter_id = raw["id"] # Ingen type-fejl
if is_heat_telegram(raw):
# Her ved type-checker: raw er dict med heat-felter
value_mwh = raw["value_mwh"]
await store_heat_reading(meter_id, value_mwh)
Self — self-refererende return type
# server/src/db/models.py
# Self til builder-pattern og fluent interface
from typing_extensions import Self
from sqlmodel import SQLModel, Field
from decimal import Decimal
import uuid
class PropertyConfigBuilder:
"""Builder til PropertyConfig — Self giver korrekt type ved arv."""
def __init__(self):
self._config: dict = {
"fixed_share": Decimal("0.30"),
"distribution_method": "hca",
"hca_correction_enabled": False,
}
def with_fixed_share(self, share: Decimal) -> Self:
if not Decimal("0") <= share <= Decimal("1"):
raise ValueError(f"Fixed share skal være 0-1: {share}")
self._config["fixed_share"] = share
return self
def with_distribution_method(self, method: str) -> Self:
valid = {"hca", "direct_heat_meter", "area_prorate"}
if method not in valid:
raise ValueError(f"Ugyldig metode: {method}. Vælg: {valid}")
self._config["distribution_method"] = method
return self
def with_hca_corrections(self, enabled: bool = True) -> Self:
self._config["hca_correction_enabled"] = enabled
return self
def build(self) -> dict:
return self._config.copy()
# Brug — fluent API:
config = (
PropertyConfigBuilder()
.with_fixed_share(Decimal("0.30"))
.with_distribution_method("hca")
.with_hca_corrections(enabled=True)
.build()
)
# Self i SQLModel til copy-with-changes mønster:
class Reading(SQLModel):
meter_installation_id: uuid.UUID
value: Decimal
unit: str
def with_value(self, new_value: Decimal) -> Self:
"""Returnér kopi med ny værdi — Self sikrer korrekt subtype."""
return self.model_copy(update={"value": new_value})
Required og NotRequired — TypedDict fintuning
# gateway/src/mqtt/types.py
# Finkornet TypedDict med Required/NotRequired
from typing import TypedDict
from typing_extensions import Required, NotRequired
class MqttPayload(TypedDict, total=False):
"""
MQTT payload-struktur.
total=False: Alle felter er optional som default.
Required/NotRequired overskriver pr. felt.
"""
# Obligatoriske felter (altid tilstede):
gateway_id: Required[str]
timestamp: Required[str]
readings: Required[list[dict]]
# Valgfrie felter:
firmware_version: NotRequired[str]
battery_levels: NotRequired[dict[str, int]]
rssi_levels: NotRequired[dict[str, int]]
gateway_uptime_seconds: NotRequired[int]
class AlarmPayload(TypedDict):
"""Alarm-payload fra gateway."""
gateway_id: str
alarm_type: str
severity: str # "critical", "warning", "info"
message: str
meter_id: NotRequired[str] # Kun ved måler-specifikke alarmer
timestamp: str
# Type-checker advarer hvis obligatoriske felter mangler:
def create_payload(gateway_id: str) -> MqttPayload:
return {
"gateway_id": gateway_id, # Required → mangler den, fejl
"timestamp": "2026-05-24T06:00:00Z",
"readings": [],
# firmware_version er NotRequired → OK at udelade
}
Unpack og TypeVarTuple — variadic generics
# server/src/api/utils.py
# Variadic type hints til tuple-baserede returnværdier
from typing_extensions import TypeVarTuple, Unpack
from typing import TypeVar
Ts = TypeVarTuple("Ts")
def validate_all(*validators: Unpack[tuple[callable, ...]]) -> list[str]:
"""
Kør alle validators og saml fejlmeddelelser.
Variadic: Accepterer vilkårligt antal validators.
"""
errors = []
for validator in validators:
result = validator()
if result:
errors.append(result)
return errors
# Mere praktisk eksempel — tuple-unpack i return:
from typing import NamedTuple
class DistributionResult(NamedTuple):
fixed_amount: Decimal
variable_amount: Decimal
total: Decimal
balance: Decimal
def unpack_distribution(result: DistributionResult):
fixed, variable, total, balance = result # Type-safe unpack
...
override — eksplicit metode-override
# server/src/repositories/base.py
# override markerer eksplicit at metode overskriver en abstrakt
from typing_extensions import override
from abc import ABC, abstractmethod
import uuid
class BaseRepository[T](ABC):
"""Abstrakt base repository."""
@abstractmethod
async def get(self, id: uuid.UUID) -> T | None: ...
@abstractmethod
async def create(self, data: dict) -> T: ...
@abstractmethod
async def delete(self, id: uuid.UUID) -> bool: ...
class PropertyRepository(BaseRepository):
"""Ejendoms-repository — override ekspliciterer interface."""
@override
async def get(self, id: uuid.UUID) -> Property | None:
async with get_session() as session:
return await session.get(Property, id)
@override
async def create(self, data: dict) -> Property:
async with get_session() as session:
prop = Property(**data)
session.add(prop)
await session.commit()
return prop
@override
async def delete(self, id: uuid.UUID) -> bool:
# Soft delete — ikke rigtig sletning:
async with get_session() as session:
prop = await session.get(Property, id)
if not prop:
return False
prop.deleted_at = datetime.utcnow()
await session.commit()
return True
# NB: Staver man forkert (f.eks. "delet" i stedet for "delete"):
# @override
# async def delet(self, ...): # Type-checker fejl: ingen base-metode matcher
Konklusion
typing_extensions giver Python 3.11 adgang til moderne type-features. TypeAlias forbedrer læsbarheden af domæne-specifikke typer (AmountDKK, TenantId). TypeGuard giver type-sikker runtime-validering af wM-Bus telegram-struktur. Self sikrer korrekte return-typer i builder-patterns og SQLModel-kopi-metoder. Required/NotRequired i TypedDict giver præcis kontrol over obligatoriske vs. valgfrie MQTT-payload-felter. @override markerer eksplicit metode-overskrivning og fanger stavefejl ved compile-time.