refacto: reorganize files to easily add an API service

server/app/adapters/mqtt.py

@@ -0,0 +1,65 @@
+import json
+import logging
+from typing import Callable
+
+import paho.mqtt.client as mqtt
+from paho.mqtt.client import ConnectFlags
+from paho.mqtt.enums import CallbackAPIVersion
+from paho.mqtt.properties import Properties
+from paho.mqtt.reasoncodes import ReasonCode
+
+from domain.exceptions import MessageBrokerError, InfrastructureError
+from domain.entities import UplinkEvent
+from ports import MessageBroker
+
+log = logging.getLogger(__name__)
+
+
+class PahoMqttBroker(MessageBroker):
+    def __init__(self, host: str, port: int, topic: str):
+        self._host = host
+        self._port = port
+        self._topic = topic
+
+    def start(self, on_uplink: Callable[[UplinkEvent], None]) -> None:
+        _client = mqtt.Client(CallbackAPIVersion.VERSION2)
+
+        def _on_connect(
+            client: mqtt.Client,
+            userdata: None,
+            flags: ConnectFlags,
+            reason_code: ReasonCode,
+            props: Properties | None,
+        ) -> None:
+            log.info("MQTT connecté (code=%s)", reason_code)
+            client.subscribe(self._topic, qos=1)
+
+        def _on_message(client: mqtt.Client, userdata: None, msg: mqtt.MQTTMessage) -> None:
+            try:
+                body = json.loads(msg.payload)
+            except json.JSONDecodeError as e:
+                log.error("Payload JSON invalide sur %s : %s", msg.topic, e)
+                return
+
+            try:
+                dev_eui = body["deviceInfo"]["devEui"]
+                pulse_count = int(body["object"]["pulse_count"])
+            except (KeyError, ValueError, TypeError) as e:
+                log.error("Champs manquants ou invalides : %s | body=%s", e, body)
+                return
+            
+            try:
+                on_uplink(UplinkEvent(dev_eui=dev_eui, pulse_count=pulse_count))
+            except InfrastructureError as e:
+                log.error("Erreur infrastructure pour %s : %s", dev_eui, e)
+
+        _client.on_connect = _on_connect
+        _client.on_message = _on_message
+        _client.reconnect_delay_set(min_delay=1, max_delay=30)
+
+        try:
+            _client.connect(self._host, self._port, keepalive=60)
+        except OSError as e:
+            raise MessageBrokerError(f"Impossible de joindre {self._host}:{self._port} : {e}") from e
+        
+        _client.loop_forever()

server/app/adapters/postgres.py

@@ -0,0 +1,63 @@
+import logging
+import time
+
+import psycopg2
+from psycopg2.extensions import connection
+from ports import DeviceRepository, ReadingRepository
+from domain.exceptions import DatabaseConnectionError, DatabaseError
+
+log = logging.getLogger(__name__)
+
+def connect(uri: str, retries: int = 10, base_delay: float = 1.0) -> connection:
+    for attempt in range(retries):
+        try:
+            conn = psycopg2.connect(uri)
+            conn.autocommit = True
+            log.info("PostgreSQL connecté")
+            return conn
+        except psycopg2.OperationalError as e:
+            delay = min(base_delay * 2 ** attempt, 30.0)
+            log.warning("Attente PostgreSQL (tentative %d/%d) : %s", attempt + 1, retries, e)
+            time.sleep(delay)
+    raise DatabaseConnectionError(f"Impossible de se connecter après {retries} tentatives")
+
+
+class PgDeviceRepository(DeviceRepository):
+    def __init__(self, conn: connection):
+        self._conn = conn
+
+    def get_or_create_device_id(self, dev_eui: str) -> str:
+        try:
+            with self._conn.cursor() as cur:
+                cur.execute(
+                    """
+                    INSERT INTO device (device_eui)
+                    VALUES (%s)
+                    ON CONFLICT (device_eui) DO NOTHING
+                """,
+                    (dev_eui,),
+                )
+                cur.execute(
+                    "SELECT device_id FROM device WHERE device_eui = %s", (dev_eui,)
+                )
+                return str(cur.fetchone()[0])
+        except psycopg2.DatabaseError as e:
+            raise DatabaseError(f"Erreur de création du device {dev_eui}") from e
+
+
+class PgReadingRepository(ReadingRepository):
+    def __init__(self, conn: connection):
+        self._conn = conn
+
+    def insert_reading(self, device_id: str, pulse_count: int) -> None:
+        try:
+            with self._conn.cursor() as cur:
+                cur.execute(
+                    """
+                    INSERT INTO reading (device_id, date, pulses)
+                    VALUES (%s, NOW(), %s)
+                """,
+                    (device_id, pulse_count),
+                )
+        except psycopg2.DatabaseError as e:
+            raise DatabaseError(f"Erreur d'enregistrement de la télérelève sur le device {device_id}") from e