Tri non déterministe : `latest_for_site` départage désormais les égalités de timestamp par `reading_id` décroissant, comme `list_history`. `uq_reading_source` autorise deux lignes au même `site_id`+`timestamp` quand la `source` diffère, donc le `LIMIT 1` pouvait renvoyer l'une ou l'autre d'un appel à l'autre. Tests : trois tests `integration` sur `latest_for_site` (plus récente, égalité de timestamp, isolation par site). Le test d'égalité échoue sans le correctif ci-dessus. Duplication : `DataQuality` et le repli vers `critical` sortent dans `app/services/data_quality.py`, partagé par `stats.py`, `site.py` et `sensor.py`, qui en portaient trois copies indépendantes. Supprime au passage deux `# type: ignore[assignment]`.
19 lines
811 B
Python
19 lines
811 B
Python
# Contrainte : `ck_reading_quality` accepte NULL et quatre valeurs seulement, alors que le contrat
|
|
# frontend n'a aucune valeur pour l'absence de qualité. `qualite_ou_critique()` replie donc sur
|
|
# `critical`, la seule des quatre qui n'induise pas une confiance qu'on n'a pas. `QUALITES_CONNUES`
|
|
# reste exposé pour les appelants qui doivent distinguer un `critical` stocké d'un repli.
|
|
|
|
from typing import Literal, get_args
|
|
|
|
DataQuality = Literal["good", "partial", "degraded", "critical"]
|
|
|
|
QUALITES_CONNUES: frozenset[str] = frozenset(get_args(DataQuality))
|
|
|
|
_PAR_VALEUR: dict[str, DataQuality] = {valeur: valeur for valeur in get_args(DataQuality)}
|
|
|
|
|
|
def qualite_ou_critique(valeur: str | None) -> DataQuality:
|
|
if valeur is None:
|
|
return "critical"
|
|
return _PAR_VALEUR.get(valeur, "critical")
|