Sensor Client (Strategy Pattern)

Overview

Pattern Strategy pour abstraire l'accès à un capteur externe : une classe de base abstraite, une implémentation réelle (HTTP avec cache), et une implémentation simulée pour les tests/développement.

File Structure

code

src/cuve-api/
├── app/
│   ├── cuve.py          # Clients capteur (abstrait, réel, simulé)

Implementation Pattern

Dataclass pour les lectures

python

from dataclasses import dataclass

@dataclass
class CuveReading:
    distance_cm: int
    timestamp: str
    ip: str
    fetched_at_epoch: float

Classe de base abstraite

python

class CuveClient:
    async def get_reading(self, force_refresh: bool = False) -> CuveReading:
        raise NotImplementedError

Client réel avec cache

python

import httpx
import time

class RealCuveClient(CuveClient):
    def __init__(self, sensor_url: str, cache_ttl_seconds: int, http_timeout_seconds: float):
        self.sensor_url = sensor_url
        self.cache_ttl_seconds = cache_ttl_seconds
        self.http_timeout_seconds = http_timeout_seconds
        self._cache: Optional[CuveReading] = None

    def _cache_valid(self) -> bool:
        if not self._cache:
            return False
        return (time.time() - self._cache.fetched_at_epoch) < self.cache_ttl_seconds

    async def get_reading(self, force_refresh: bool = False) -> CuveReading:
        if (not force_refresh) and self._cache_valid():
            return self._cache

        async with httpx.AsyncClient(timeout=self.http_timeout_seconds) as client:
            resp = await client.get(self.sensor_url)
            resp.raise_for_status()
            data = resp.json()

        reading = CuveReading(
            distance_cm=int(data.get("distance_cm", -1)),
            timestamp=str(data.get("timestamp", "")),
            ip=str(data.get("ip", "")),
            fetched_at_epoch=time.time(),
        )
        self._cache = reading
        return reading

Client simulé

python

import random

class SimCuveClient(CuveClient):
    def __init__(self, base_distance_cm: int = 30):
        self.base = base_distance_cm
        self._last = base_distance_cm

    async def get_reading(self, force_refresh: bool = False) -> CuveReading:
        drift = random.choice([-1, 0, 0, 0, 1])
        noise = random.randint(-2, 2)
        self._last = max(0, self._last + drift + noise)

        ts = time.strftime("%Y-%m-%dT%H:%M:%S%z", time.localtime())

        return CuveReading(
            distance_cm=int(self._last),
            timestamp=ts,
            ip="simulated",
            fetched_at_epoch=time.time(),
        )

Sélection du client selon le mode

python

# Dans main.py
if settings.mode == "sim":
    cuve = SimCuveClient(base_distance_cm=30)
else:
    cuve = RealCuveClient(
        sensor_url=settings.sensor_url,
        cache_ttl_seconds=settings.cache_ttl_seconds,
        http_timeout_seconds=settings.http_timeout_seconds,
    )

Rules

Do

•Utiliser une classe abstraite pour définir l'interface
•Implémenter le cache avec TTL pour réduire les appels réseau
•Utiliser httpx.AsyncClient avec context manager pour les requêtes async
•Configurer un timeout explicite pour les appels HTTP
•Logger les lectures invalides sans lever d'exception
•Permettre force_refresh pour bypasser le cache

Don't

•Ne pas réutiliser un AsyncClient entre les requêtes (utiliser context manager)
•Ne pas ignorer silencieusement les erreurs HTTP (utiliser raise_for_status())
•Ne pas hardcoder les URLs ou timeouts

File Location

•Clients capteur : src/cuve-api/app/cuve.py