import logging import time import psycopg2 from psycopg2.extensions import connection from ports import DeviceRepository, ReadingRepository from 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