Python Abstract Base Classes — ABC og abstractmethod mønstre

ABC og abstractmethod fra Python's abc-modul håndhæver interfaces ved instansiering. Her er de praktiske mønstre til IoT-platform og repository pattern.

---

ABC grundlæggende

from abc import ABC, abstractmethod
from typing import AsyncIterator
from datetime import datetime
import uuid


class MeterReader(ABC):
    """Abstract base class for alle måler-læsere."""

    @abstractmethod
    async def read(self, meter_id: str) -> float:
        """Læs aktuelt forbrug fra måler."""
        ...

    @abstractmethod
    async def read_history(
        self,
        meter_id: str,
        since: datetime,
    ) -> AsyncIterator[tuple[datetime, float]]:
        """Stream historiske aflæsninger."""
        ...

    @property
    @abstractmethod
    def protocol(self) -> str:
        """Protokol-identifikator (fx 'wmbus', 'modbus')."""
        ...

    def __init_subclass__(cls, **kwargs):
        """Hook: Kør validering ved subclass-definition."""
        super().__init_subclass__(**kwargs)
        if not hasattr(cls, "SUPPORTED_MANUFACTURERS"):
            raise TypeError(
                f"{cls.__name__} skal definere SUPPORTED_MANUFACTURERS"
            )


# Instansiering af ABC fejler:
reader = MeterReader()    # TypeError: Can't instantiate abstract class

---

Abstract repository pattern

# server/src/repositories/base.py

from abc import ABC, abstractmethod
from typing import TypeVar, Generic
import uuid

T = TypeVar("T")


class BaseRepository(ABC, Generic[T]):
    """
    Abstract repository — definerer CRUD-interface.
    Konkrete implementeringer: SQL, mock, cache-wrapper.
    """

    @abstractmethod
    async def get(self, id: uuid.UUID) -> T | None:
        ...

    @abstractmethod
    async def get_by_tenant(
        self,
        tenant_id: uuid.UUID,
        limit: int = 100,
        offset: int = 0,
    ) -> list[T]:
        ...

    @abstractmethod
    async def create(self, obj: T) -> T:
        ...

    @abstractmethod
    async def update(self, obj: T) -> T:
        ...

    @abstractmethod
    async def soft_delete(self, id: uuid.UUID, deleted_by: uuid.UUID) -> bool:
        ...

    @abstractmethod
    async def count(self, tenant_id: uuid.UUID) -> int:
        ...


# Konkret SQL-implementering:
class SQLReadingRepository(BaseRepository[Reading]):
    SUPPORTED_MANUFACTURERS = ["*"]

    def __init__(self, session: AsyncSession):
        self._session = session

    async def get(self, id: uuid.UUID) -> Reading | None:
        return await self._session.get(Reading, id)

    async def get_by_tenant(
        self,
        tenant_id: uuid.UUID,
        limit: int = 100,
        offset: int = 0,
    ) -> list[Reading]:
        stmt = (
            select(Reading)
            .where(Reading.tenant_id == tenant_id)
            .where(Reading.deleted_at.is_(None))
            .limit(limit)
            .offset(offset)
        )
        return (await self._session.execute(stmt)).scalars().all()

    # ... øvrige abstract methods implementeret ...


# Mock-implementering til tests:
class InMemoryReadingRepository(BaseRepository[Reading]):
    SUPPORTED_MANUFACTURERS = ["*"]

    def __init__(self):
        self._store: dict[uuid.UUID, Reading] = {}

    async def get(self, id: uuid.UUID) -> Reading | None:
        return self._store.get(id)

    async def create(self, obj: Reading) -> Reading:
        self._store[obj.id] = obj
        return obj

    # ... øvrige abstract methods ...

---

Plugin-system med ABCMeta

# gateway/src/drivers/base.py
# Plugin-system til wM-Bus-drivere for forskellige fabrikater

from abc import ABCMeta, abstractmethod


class DriverRegistry:
    """Registry over alle tilgængelige wM-Bus-drivere."""
    _drivers: dict[str, type] = {}

    @classmethod
    def register(cls, manufacturer_id: str):
        """Decorator til at registrere driver."""
        def decorator(driver_cls):
            cls._drivers[manufacturer_id] = driver_cls
            return driver_cls
        return decorator

    @classmethod
    def get(cls, manufacturer_id: str) -> type | None:
        return cls._drivers.get(manufacturer_id)


class WmbusDriverMeta(ABCMeta):
    """
    Metaclass der auto-registrerer drivere i DriverRegistry.
    Kræver MANUFACTURER_ID class-attribut.
    """
    def __new__(mcs, name, bases, namespace, **kwargs):
        cls = super().__new__(mcs, name, bases, namespace, **kwargs)

        # Registrér kun konkrete klasser (ikke ABC'en selv):
        if hasattr(cls, "MANUFACTURER_ID") and cls.MANUFACTURER_ID:
            DriverRegistry._drivers[cls.MANUFACTURER_ID] = cls

        return cls


class WmbusDriver(ABC, metaclass=WmbusDriverMeta):
    """Abstract base class for alle wM-Bus-drivere."""
    MANUFACTURER_ID: str = ""

    @abstractmethod
    def parse(self, raw_bytes: bytes) -> dict:
        """Parser rå telegram til struktureret data."""
        ...

    @abstractmethod
    def supports_encryption(self) -> bool:
        """Understøtter denne driver AES-128 CTR?"""
        ...

    @classmethod
    @abstractmethod
    def get_wmbusmeters_driver_name(cls) -> str:
        """Returnér wmbusmeters driver-navn (fx 'apator162')."""
        ...


# Konkrete drivere registreres automatisk:
class KamstrupDriver(WmbusDriver):
    MANUFACTURER_ID = "KAM"

    def parse(self, raw_bytes: bytes) -> dict:
        # Kamstrup-specifik parsing
        return {"manufacturer": "Kamstrup", "value": ...}

    def supports_encryption(self) -> bool:
        return True

    @classmethod
    def get_wmbusmeters_driver_name(cls) -> str:
        return "multical21"


class EngelmannDriver(WmbusDriver):
    MANUFACTURER_ID = "ELS"

    def parse(self, raw_bytes: bytes) -> dict:
        return {"manufacturer": "Engelmann", "value": ...}

    def supports_encryption(self) -> bool:
        return True

    @classmethod
    def get_wmbusmeters_driver_name(cls) -> str:
        return "sensostar"


# Brug:
driver_cls = DriverRegistry.get("KAM")
if driver_cls:
    driver = driver_cls()
    data = driver.parse(raw_telegram)

---

Abstract vs. Protocol — hvornår hvad?

# ABC — brug når:
#   - Du vil håndhæve interface ved instansiering (fejl ved oprettelse)
#   - Klasser skal eksplicit arve (nominativ subtyping)
#   - Du har fælles implementeret kode i base class
#   - Plugin-system med auto-registrering via metaclass

# Protocol — brug når:
#   - Eksisterende klasser skal bruges uden at ændre dem (duck typing)
#   - Tredjepartsklasser skal satisfie interface
#   - Type-checking uden runtime overhead
#   - Test-mocks der ikke skal arve

# Eksempel på hybrid (ABC + Protocol):
from typing import Protocol, runtime_checkable

@runtime_checkable
class Readable(Protocol):
    async def read(self, meter_id: str) -> float: ...


class MeterReader(ABC):
    """ABC giver fælles kode + håndhævet interface."""

    @abstractmethod
    async def read(self, meter_id: str) -> float: ...

    async def read_with_retry(self, meter_id: str, retries: int = 3) -> float:
        """Fælles implementering i base class."""
        for attempt in range(retries):
            try:
                return await self.read(meter_id)
            except ConnectionError:
                if attempt == retries - 1:
                    raise
                await asyncio.sleep(2 ** attempt)


# Verificér at implementation satisfier Protocol:
reader = WmbusReader()
assert isinstance(reader, Readable)    # True (runtime_checkable)

---

Multiple inheritance med MRO

# Forsigtigt med multiple inheritance — brug MRO korrekt:

class Serializable(ABC):
    @abstractmethod
    def to_dict(self) -> dict: ...

    @abstractmethod
    def to_json(self) -> str: ...


class Auditable(ABC):
    @abstractmethod
    def audit_info(self) -> dict: ...


class TimestampedRepository(BaseRepository[T], Auditable):
    """Multiple inheritance — kræver alle abstract methods."""

    def audit_info(self) -> dict:
        return {
            "created_at": datetime.utcnow().isoformat(),
            "class": type(self).__name__,
        }

    # + alle BaseRepository abstract methods ...


# MRO-check:
print(TimestampedRepository.__mro__)
# [TimestampedRepository, BaseRepository, Auditable, ABC, Generic, object]

# super() følger MRO automatisk:
class CachingRepository(TimestampedRepository[T]):
    def __init__(self, inner: BaseRepository[T], cache):
        self._inner = inner
        self._cache = cache

    async def get(self, id: uuid.UUID) -> T | None:
        cached = await self._cache.get(str(id))
        if cached:
            return cached
        result = await self._inner.get(id)
        if result:
            await self._cache.set(str(id), result, ttl=300)
        return result

---

Konklusion

ABC og abstractmethod håndhæver interfaces ved instansiering — runtime fejl ved forsøg på at oprette ufuldstændig klasse. Abstract repository pattern (BaseRepository[T]) definerer CRUD-interface og muliggør mock-implementeringer i tests. Plugin-systemer med ABCMeta auto-registrerer drivere (Kamstrup, Engelmann, Diehl) ved class-definition. Brug ABC til nominativ subtyping med fælles kode; brug Protocol til duck typing af eksisterende klasser.

Se Python Protocol guide eller FastAPI dependency injection guide.