fix(backend,airflow,ml): applique les corrections de revue sur la PR #162
This commit is contained in:
Dorian
@@ -1,17 +1,19 @@
|
||||
# Contrainte : la réponse de l'API Mock est une entrée hostile, pas une source de confiance.
|
||||
# Voir OWASP API10 dans docs/architecture/owasp-traceabilite.md. Rien de ce qu'elle renvoie
|
||||
# n'atteint la base sans passer par build_site_row() ou build_reading_row() : seuls les champs
|
||||
# attendus sont recopiés, les grandeurs physiques sont bornées par PHYSICAL_BOUNDS et la taille
|
||||
# des tableaux est plafonnée par MAX_SITES et par --limit. Une valeur hors bornes devient NULL
|
||||
# et laisse sa trace dans null_reasons plutôt que de lever : le mock émet des anomalies par
|
||||
# construction, et raw_data conserve de toute façon la réponse d'origine intacte.
|
||||
# attendus sont recopiés, les grandeurs physiques sont bornées par PHYSICAL_BOUNDS, la taille des
|
||||
# tableaux est plafonnée par MAX_SITES et par limit_for_window() (dérivé de la fenêtre, jamais
|
||||
# fourni par l'appelant), et les lectures dont le timestamp déborde de la fenêtre demandée sont
|
||||
# écartées (fetch_readings). Une valeur hors bornes devient NULL et laisse sa trace dans
|
||||
# null_reasons plutôt que de lever : le mock émet des anomalies par construction, et raw_data
|
||||
# conserve de toute façon la réponse d'origine intacte.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
import json
|
||||
from datetime import datetime
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
@@ -182,6 +184,21 @@ async def upsert_sites(
|
||||
)
|
||||
|
||||
|
||||
def _timestamp_in_window(
|
||||
reading: dict[str, Any], start_time: datetime, end_time: datetime
|
||||
) -> bool:
|
||||
valeur = reading.get("timestamp")
|
||||
if not isinstance(valeur, str):
|
||||
return False
|
||||
|
||||
try:
|
||||
instant = parse_datetime(valeur)
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
return start_time <= instant < end_time
|
||||
|
||||
|
||||
async def fetch_readings(
|
||||
client: httpx.AsyncClient,
|
||||
site_id: str,
|
||||
@@ -209,7 +226,22 @@ async def fetch_readings(
|
||||
if len(payload) > limit:
|
||||
raise ValueError(f"La réponse /api/v1/readings dépasse la limite demandée de {limit}.")
|
||||
|
||||
return payload
|
||||
# Le garde-fou `refuse_if_overlaps_historical_dataset` ne vérifie que la fenêtre demandée :
|
||||
# une réponse (bug du mock, ou hostile) dont les `timestamp` débordent de
|
||||
# `[start_time, end_time)` contournerait ce contrôle et écrirait exactement le doublon
|
||||
# inter-source qu'il doit empêcher. Écarter ces lectures ici rend le contrôle par fenêtre
|
||||
# suffisant.
|
||||
dans_la_fenetre = [
|
||||
lecture
|
||||
for lecture in payload
|
||||
if isinstance(lecture, dict) and _timestamp_in_window(lecture, start_time, end_time)
|
||||
]
|
||||
|
||||
if len(dans_la_fenetre) != len(payload):
|
||||
ecartees = len(payload) - len(dans_la_fenetre)
|
||||
print(f"{site_id}: {ecartees} lecture(s) hors fenêtre écartée(s).")
|
||||
|
||||
return dans_la_fenetre
|
||||
|
||||
|
||||
def build_reading_row(
|
||||
@@ -247,8 +279,9 @@ def build_reading_row(
|
||||
|
||||
# `uq_reading_source` autorise deux lignes au même (site_id, timestamp) dès que `source` diffère :
|
||||
# sans ce garde-fou, importer une fenêtre déjà couverte par le dataset historique (source='csv')
|
||||
# dupliquerait silencieusement chaque point plutôt que de lever une erreur, et rien côté lecture
|
||||
# (pipeline ML, GET /readings) ne saurait laquelle des deux lectures retenir.
|
||||
# dupliquerait silencieusement chaque point plutôt que de lever une erreur. Ce garde-fou protège
|
||||
# l'ingestion ; il ne dit rien de la lecture (`GET /readings` renvoie les deux lignes en cas de
|
||||
# doublon malgré tout, cf. la section réconciliation de 40-data.md).
|
||||
OVERLAP_CHECK = text(
|
||||
"SELECT count(*) FROM reading WHERE source = :source_csv "
|
||||
"AND timestamp >= :start_time AND timestamp < :end_time"
|
||||
@@ -332,41 +365,41 @@ def build_reading_batch(
|
||||
return [build_reading_row(reading) for reading in readings]
|
||||
|
||||
|
||||
def limit_for_window(start_time: datetime, end_time: datetime) -> int:
|
||||
"""Nombre de lectures à demander pour que l'API Mock en rende une par heure, alignée.
|
||||
|
||||
L'API ne renvoie pas un flux à un rythme naturel : elle répartit exactement `limit` lectures,
|
||||
espacées uniformément, sur toute la fenêtre `[start_time, end_time)` demandée (vérifié
|
||||
empiriquement). `limit` = nombre d'heures de la fenêtre est donc le seul réglage cohérent avec
|
||||
le grain horaire du reste du schéma (`period_minutes=60`, historique CSV à une ligne/heure) ;
|
||||
un `limit` plus grand fabriquerait des lectures infra-horaires, incompatibles avec les lags
|
||||
positionnels de `build_features`. La fenêtre doit donc couvrir un nombre entier d'heures.
|
||||
"""
|
||||
duree = end_time - start_time
|
||||
heures, reste = divmod(duree.total_seconds(), 3600)
|
||||
|
||||
if reste != 0:
|
||||
raise ValueError(
|
||||
"La fenêtre doit couvrir un nombre entier d'heures pour obtenir une lecture par "
|
||||
f"heure alignée : [{start_time.isoformat()}, {end_time.isoformat()}) n'en couvre pas."
|
||||
)
|
||||
|
||||
if heures > MAX_LIMIT:
|
||||
raise ValueError(
|
||||
f"La fenêtre demandée couvre {int(heures)}h, au-delà du plafond de {MAX_LIMIT} "
|
||||
"lectures accepté par l'API Mock."
|
||||
)
|
||||
|
||||
return int(heures)
|
||||
|
||||
|
||||
async def import_mock_api_history(
|
||||
start_time: datetime,
|
||||
end_time: datetime,
|
||||
limit: int,
|
||||
dry_run: bool,
|
||||
) -> None:
|
||||
settings = get_settings()
|
||||
|
||||
async with create_mock_api_client() as client:
|
||||
sites = await fetch_sites(client)
|
||||
|
||||
print(f"Sites récupérés : {len(sites)}")
|
||||
|
||||
all_readings: list[dict[str, Any]] = []
|
||||
|
||||
for site in sites:
|
||||
site_id = read_text(site, "site_id")
|
||||
|
||||
readings = await fetch_readings(
|
||||
client=client,
|
||||
site_id=site_id,
|
||||
start_time=start_time,
|
||||
end_time=end_time,
|
||||
limit=limit,
|
||||
)
|
||||
|
||||
print(f"{site_id}: {len(readings)} lectures")
|
||||
|
||||
all_readings.extend(readings)
|
||||
|
||||
print(f"Lectures récupérées : {len(all_readings)}")
|
||||
|
||||
if dry_run:
|
||||
print("Dry-run terminé : aucune donnée écrite.")
|
||||
return
|
||||
limit = limit_for_window(start_time, end_time)
|
||||
|
||||
engine = create_async_engine(
|
||||
str(settings.database_url),
|
||||
@@ -374,9 +407,40 @@ async def import_mock_api_history(
|
||||
)
|
||||
|
||||
try:
|
||||
async with engine.begin() as connection:
|
||||
# Garde-fou d'abord, y compris en dry-run : il est en lecture seule, et annoncer un
|
||||
# succès pour une fenêtre que l'import réel refusera serait trompeur.
|
||||
async with engine.connect() as connection:
|
||||
await refuse_if_overlaps_historical_dataset(connection, start_time, end_time)
|
||||
|
||||
async with create_mock_api_client() as client:
|
||||
sites = await fetch_sites(client)
|
||||
|
||||
print(f"Sites récupérés : {len(sites)}")
|
||||
|
||||
all_readings: list[dict[str, Any]] = []
|
||||
|
||||
for site in sites:
|
||||
site_id = read_text(site, "site_id")
|
||||
|
||||
readings = await fetch_readings(
|
||||
client=client,
|
||||
site_id=site_id,
|
||||
start_time=start_time,
|
||||
end_time=end_time,
|
||||
limit=limit,
|
||||
)
|
||||
|
||||
print(f"{site_id}: {len(readings)} lectures")
|
||||
|
||||
all_readings.extend(readings)
|
||||
|
||||
print(f"Lectures récupérées : {len(all_readings)}")
|
||||
|
||||
if dry_run:
|
||||
print("Dry-run terminé : aucune donnée écrite.")
|
||||
return
|
||||
|
||||
async with engine.begin() as connection:
|
||||
await upsert_sites(
|
||||
connection,
|
||||
sites,
|
||||
@@ -397,7 +461,14 @@ async def import_mock_api_history(
|
||||
|
||||
|
||||
def parse_datetime(value: str) -> datetime:
|
||||
return datetime.fromisoformat(value.replace("Z", "+00:00"))
|
||||
# Sans fuseau, l'API le traite comme reçu, telle quelle, mais l'encodeur `timestamptz`
|
||||
# d'asyncpg lirait un datetime naif dans le fuseau *local du processus* (correct dans le
|
||||
# conteneur Airflow en UTC, décalé de 1-2h pour un import manuel lancé depuis un poste en
|
||||
# Europe/Paris). Poser `tzinfo=UTC` explicitement, même pattern que `_vers_utc()` dans
|
||||
# `app/services/reading.py`, garantit que la borne envoyée à l'API et celle comparée en SQL
|
||||
# (refuse_if_overlaps_historical_dataset) désignent le même instant.
|
||||
instant = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
||||
return instant if instant.tzinfo is not None else instant.replace(tzinfo=UTC)
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
@@ -415,12 +486,6 @@ def parse_args() -> argparse.Namespace:
|
||||
type=parse_datetime,
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--limit",
|
||||
type=int,
|
||||
default=MAX_LIMIT,
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--dry-run",
|
||||
action="store_true",
|
||||
@@ -432,9 +497,6 @@ def parse_args() -> argparse.Namespace:
|
||||
def main() -> None:
|
||||
args = parse_args()
|
||||
|
||||
if args.limit < 1 or args.limit > MAX_LIMIT:
|
||||
raise ValueError(f"--limit doit être compris entre 1 et {MAX_LIMIT}.")
|
||||
|
||||
if args.start_time >= args.end_time:
|
||||
raise ValueError("--start-time doit être antérieur à --end-time.")
|
||||
|
||||
@@ -442,7 +504,6 @@ def main() -> None:
|
||||
import_mock_api_history(
|
||||
start_time=args.start_time,
|
||||
end_time=args.end_time,
|
||||
limit=args.limit,
|
||||
dry_run=args.dry_run,
|
||||
)
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user