SQLite Repository

Overview

Pattern pour la persistance avec SQLite : dataclasses frozen pour les entités, fonctions de requête avec gestion explicite des connexions, et initialisation du schéma.

File Structure

code

src/cuve-api/
├── app/
│   ├── db.py            # Repository SQLite

Implementation Pattern

Entité avec dataclass frozen

python

from dataclasses import dataclass
from typing import Optional, List, Literal

Order = Literal["max", "min"]
Period = Literal["day", "week", "month", "year", "all"]

@dataclass(frozen=True)
class DbReading:
    id: int
    distance_cm: int
    sensor_timestamp: str
    sensor_ip: str
    fetched_at_epoch: float

Initialisation du schéma

python

def init_db(db_path: str) -> None:
    con = sqlite3.connect(db_path)
    try:
        con.execute("""
            CREATE TABLE IF NOT EXISTS readings (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                distance_cm INTEGER NOT NULL,
                sensor_timestamp TEXT NOT NULL,
                sensor_ip TEXT NOT NULL,
                fetched_at_epoch REAL NOT NULL
            )
        """)
        con.execute("CREATE INDEX IF NOT EXISTS idx_readings_fetched_at ON readings(fetched_at_epoch)")
        con.execute("CREATE INDEX IF NOT EXISTS idx_readings_distance ON readings(distance_cm)")
        con.commit()
    finally:
        con.close()

Connexion avec row_factory

python

def _connect(db_path: str) -> sqlite3.Connection:
    con = sqlite3.connect(db_path)
    con.row_factory = sqlite3.Row
    return con

Lecture (SELECT)

python

def get_last(db_path: str) -> Optional[DbReading]:
    con = _connect(db_path)
    try:
        row = con.execute("""
            SELECT id, distance_cm, sensor_timestamp, sensor_ip, fetched_at_epoch
            FROM readings
            ORDER BY fetched_at_epoch DESC, id DESC
            LIMIT 1
        """).fetchone()
        if not row:
            return None
        return DbReading(**dict(row))
    finally:
        con.close()

Insertion avec dédoublonnage

python

def insert_reading(
        db_path: str,
        distance_cm: int,
        sensor_timestamp: str,
        sensor_ip: str,
        fetched_at_epoch: float,
        dedupe_by_sensor_ts: bool = True,
) -> bool:
    """Retourne True si inséré, False si ignoré (doublon)."""
    con = _connect(db_path)
    try:
        if dedupe_by_sensor_ts:
            row = con.execute("""
                SELECT sensor_timestamp FROM readings
                ORDER BY fetched_at_epoch DESC, id DESC LIMIT 1
            """).fetchone()
            if row and row["sensor_timestamp"] == sensor_timestamp:
                return False

        con.execute("""
            INSERT INTO readings(distance_cm, sensor_timestamp, sensor_ip, fetched_at_epoch)
            VALUES(?, ?, ?, ?)
        """, (distance_cm, sensor_timestamp, sensor_ip, fetched_at_epoch))
        con.commit()
        return True
    finally:
        con.close()

Requêtes avec filtres dynamiques

python

def get_extremes(db_path: str, period: Period, n: int, order: Order) -> List[DbReading]:
    since = _since_epoch(period)
    con = _connect(db_path)
    try:
        where = ""
        params = []
        if since is not None:
            where = "WHERE fetched_at_epoch >= ?"
            params.append(since)

        order_clause = "DESC" if order == "max" else "ASC"
        sql = f"""
            SELECT id, distance_cm, sensor_timestamp, sensor_ip, fetched_at_epoch
            FROM readings {where}
            ORDER BY distance_cm {order_clause}, fetched_at_epoch DESC
            LIMIT ?
        """
        params.append(n)
        rows = con.execute(sql, tuple(params)).fetchall()
        return [DbReading(**dict(r)) for r in rows]
    finally:
        con.close()

Rules

Do

•Utiliser dataclass(frozen=True) pour les entités (immutabilité)
•Toujours fermer la connexion dans un bloc finally
•Utiliser row_factory = sqlite3.Row pour accéder aux colonnes par nom
•Créer des index sur les colonnes fréquemment filtrées/triées
•Utiliser des paramètres ? pour éviter les injections SQL
•Retourner Optional[T] ou List[T] explicitement

Don't

•Ne pas garder de connexion globale ouverte
•Ne pas construire des requêtes par concaténation de strings utilisateur
•Ne pas oublier le commit() après INSERT/UPDATE/DELETE

File Location

•Repository : src/cuve-api/app/db.py