diff --git a/.github/workflows/airflow.yml b/.github/workflows/airflow.yml index d6d314a..66be8b8 100644 --- a/.github/workflows/airflow.yml +++ b/.github/workflows/airflow.yml @@ -74,11 +74,12 @@ jobs: # `--help` sort par argparse avant `get_settings()` : ni base ni secret requis, et # l'import des modules prouve que l'environnement /opt/backend est complet. - - name: Vérifie que les quatre commandes backend s'importent sans réseau + - name: Vérifie que les cinq commandes backend s'importent sans réseau run: > docker run --rm --network none enervision-airflow:ci bash -c "cd /opt/backend && env -u VIRTUAL_ENV uv run --no-sync python -m app.detection.internal_alerts --help && env -u VIRTUAL_ENV uv run --no-sync python -m app.cli generate-recommendations --help && env -u VIRTUAL_ENV uv run --no-sync python -m app.etl.historical_import --help - && env -u VIRTUAL_ENV uv run --no-sync python -m app.etl.mock_api_import --help" + && env -u VIRTUAL_ENV uv run --no-sync python -m app.etl.mock_api_import --help + && env -u VIRTUAL_ENV uv run --no-sync python -m app.etl.reading_retention --help" diff --git a/apps/backend/app/core/config.py b/apps/backend/app/core/config.py index be3a63b..a0f591f 100644 --- a/apps/backend/app/core/config.py +++ b/apps/backend/app/core/config.py @@ -76,9 +76,25 @@ class Settings(BaseSettings): expose_api_docs: bool | None = None metrics_token: SecretStr | None = None - # Compose passe `APP_METRICS_TOKEN` vide quand aucun jeton n'est posé : vide vaut absent, sinon - # `/metrics` exigerait un `Bearer` sans valeur et plus rien ne pourrait le scruter. - @field_validator("metrics_token", mode="before") + s3_endpoint_url: str | None = None + s3_region: str = "garage" + s3_access_key: str | None = None + s3_secret_key: SecretStr | None = None + s3_bucket: str | None = None + s3_sse_key: SecretStr | None = None + reading_retention_days: int = Field(default=1095, ge=30) + + # Compose passe `APP_METRICS_TOKEN` et les réglages S3 vides quand rien n'est posé : vide vaut + # absent, sinon `/metrics` exigerait un `Bearer` sans valeur et l'archivage un endpoint vide. + @field_validator( + "metrics_token", + "s3_endpoint_url", + "s3_access_key", + "s3_secret_key", + "s3_bucket", + "s3_sse_key", + mode="before", + ) @classmethod def _jeton_vide_vaut_absent(cls, valeur: object) -> object: return None if valeur == "" else valeur diff --git a/apps/backend/app/etl/reading_retention.py b/apps/backend/app/etl/reading_retention.py new file mode 100644 index 0000000..f634702 --- /dev/null +++ b/apps/backend/app/etl/reading_retention.py @@ -0,0 +1,349 @@ +# Pourquoi : la suppression n'est pas confiée à add_retention_policy, qui ignorerait l'export. +# archive_reading_chunks() exporte chaque chunk vers Garage, le relit, puis le supprime seul. +# Piège : drop_chunks pose un verrou exclusif sur reading, site et dataset jusqu'au COMMIT. La +# suppression tient donc dans une transaction dédiée et courte, séparée de la lecture du chunk. + +from __future__ import annotations + +import argparse +import asyncio +import base64 +import hashlib +import io +import json +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from typing import TYPE_CHECKING, Any + +import anyio.to_thread +import boto3 +import pandas as pd +from botocore.exceptions import ClientError +from pydantic import SecretStr +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine, create_async_engine + +from app.core.config import Settings, get_settings + +if TYPE_CHECKING: + from types_boto3_s3.client import S3Client + +SSE_KEY_LENGTH = 32 +FORMAT_BORNE = "%Y%m%dT%H%M%SZ" + +ELIGIBLE_CHUNKS = text( + "SELECT chunk_schema, chunk_name, range_start, range_end " + "FROM timescaledb_information.chunks " + "WHERE hypertable_name = 'reading' AND range_end <= :older_than " + "ORDER BY range_start" +) + +# Lecture via l'hypertable, jamais la table interne : l'exclusion de partition vise le seul chunk. +CHUNK_ROWS = text( + "SELECT * FROM reading WHERE timestamp >= :start AND timestamp < :end " + "ORDER BY timestamp, reading_id" +) + +# Les deux bornes sont inclusives pour drop_chunks : celles du chunk le désignent, et lui seul. +DROP_CHUNK = text( + "SELECT drop_chunks('reading', " + "older_than => CAST(:end AS timestamptz), newer_than => CAST(:start AS timestamptz))" +) + + +@dataclass(frozen=True) +class Chunk: + schema: str + name: str + range_start: datetime + range_end: datetime + + @property + def qualified_name(self) -> str: + return f"{self.schema}.{self.name}" + + +@dataclass +class Rapport: + chunks_vus: int = 0 + exportes: int = 0 + deja_presents: int = 0 + supprimes: int = 0 + lignes: int = 0 + + +def object_key(chunk: Chunk) -> str: + start = chunk.range_start.astimezone(UTC) + end = chunk.range_end.astimezone(UTC) + return ( + f"reading/{start.year}/reading_{start.strftime(FORMAT_BORNE)}_" + f"{end.strftime(FORMAT_BORNE)}.csv.gz" + ) + + +async def eligible_chunks(conn: AsyncConnection, older_than: datetime) -> list[Chunk]: + result = await conn.execute(ELIGIBLE_CHUNKS, {"older_than": older_than}) + return [ + Chunk( + schema=row["chunk_schema"], + name=row["chunk_name"], + range_start=row["range_start"], + range_end=row["range_end"], + ) + for row in result.mappings().all() + ] + + +async def read_chunk_rows(conn: AsyncConnection, chunk: Chunk) -> list[dict[str, Any]]: + result = await conn.execute(CHUNK_ROWS, {"start": chunk.range_start, "end": chunk.range_end}) + return [dict(row) for row in result.mappings().all()] + + +def _csv_cell(value: object) -> object: + if isinstance(value, dict | list): + return json.dumps(value, ensure_ascii=False, sort_keys=True) + return value + + +def serialize_csv_gzip(rows: list[dict[str, Any]]) -> bytes: + if not rows: + raise ValueError("Aucune ligne à sérialiser : un CSV sans colonne ne se relit pas.") + + frame = pd.DataFrame([{name: _csv_cell(value) for name, value in row.items()} for row in rows]) + buffer = io.BytesIO() + frame.to_csv(buffer, mode="wb", index=False, compression={"method": "gzip", "mtime": 0}) + return buffer.getvalue() + + +def sha256_of(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def _is_missing_object(erreur: ClientError) -> bool: + error = erreur.response.get("Error") + metadata = erreur.response.get("ResponseMetadata") + code = error.get("Code") if error is not None else None + status = metadata.get("HTTPStatusCode") if metadata is not None else None + return code == "NoSuchKey" or status == 404 + + +class ArchiveStore: + def __init__(self, client: S3Client, bucket: str, sse_key: bytes | None) -> None: + self._client = client + self._bucket = bucket + self._sse_key = sse_key + + # boto3 encode lui-même la clé en base64 et calcule son MD5 : la fournir brute, sans MD5. + def _sse_headers(self) -> dict[str, Any]: + if self._sse_key is None: + return {} + return {"SSECustomerAlgorithm": "AES256", "SSECustomerKey": self._sse_key} + + def put(self, key: str, body: bytes, metadata: dict[str, str]) -> None: + self._client.put_object( + Bucket=self._bucket, + Key=key, + Body=body, + ContentType="text/csv", + ContentEncoding="gzip", + Metadata=metadata, + **self._sse_headers(), + ) + + def fetch_sha256(self, key: str) -> str | None: + try: + response = self._client.get_object(Bucket=self._bucket, Key=key, **self._sse_headers()) + except ClientError as erreur: + if _is_missing_object(erreur): + return None + raise + return sha256_of(response["Body"].read()) + + +def decode_sse_key(encoded: SecretStr | None) -> bytes | None: + if encoded is None: + return None + + key = base64.b64decode(encoded.get_secret_value(), validate=True) + if len(key) != SSE_KEY_LENGTH: + raise ValueError( + f"APP_S3_SSE_KEY doit encoder exactement {SSE_KEY_LENGTH} octets en base64, " + f"pas {len(key)}." + ) + return key + + +def build_archive_store(settings: Settings) -> ArchiveStore: + endpoint = settings.s3_endpoint_url + access_key = settings.s3_access_key + secret_key = settings.s3_secret_key + bucket = settings.s3_bucket + + if endpoint is None or access_key is None or secret_key is None or bucket is None: + raise ValueError( + "L'archivage vers Garage exige APP_S3_ENDPOINT_URL, APP_S3_ACCESS_KEY, " + "APP_S3_SECRET_KEY et APP_S3_BUCKET." + ) + + client = boto3.client( + "s3", + endpoint_url=endpoint, + aws_access_key_id=access_key, + aws_secret_access_key=secret_key.get_secret_value(), + region_name=settings.s3_region, + ) + return ArchiveStore(client, bucket=bucket, sse_key=decode_sse_key(settings.s3_sse_key)) + + +async def drop_chunk(conn: AsyncConnection, chunk: Chunk) -> None: + result = await conn.execute(DROP_CHUNK, {"start": chunk.range_start, "end": chunk.range_end}) + supprimes = list(result.scalars().all()) + + if supprimes != [chunk.qualified_name]: + raise RuntimeError( + f"drop_chunks devait supprimer exactement {chunk.qualified_name}, " + f"il a rendu {supprimes}." + ) + + +async def _export( + store: ArchiveStore, + key: str, + rows: list[dict[str, Any]], + *, + dry_run: bool, + rapport: Rapport, +) -> str: + body = serialize_csv_gzip(rows) + sha = sha256_of(body) + + if await anyio.to_thread.run_sync(store.fetch_sha256, key) == sha: + rapport.deja_presents += 1 + return f"{len(body)} octets déjà présents" + + if dry_run: + return f"{len(body)} octets à exporter" + + metadata = {"sha256": sha, "rows": str(len(rows))} + await anyio.to_thread.run_sync(store.put, key, body, metadata) + relu = await anyio.to_thread.run_sync(store.fetch_sha256, key) + + if relu != sha: + raise RuntimeError( + f"Relecture de {key} : sha256 {relu} au lieu de {sha}, le chunk est conservé." + ) + + rapport.exportes += 1 + return f"{len(body)} octets exportés et relus" + + +async def _archive_chunk( + engine: AsyncEngine, + store: ArchiveStore, + chunk: Chunk, + *, + dry_run: bool, + rapport: Rapport, +) -> None: + async with engine.connect() as conn: + rows = await read_chunk_rows(conn, chunk) + + key = object_key(chunk) + rapport.lignes += len(rows) + + if rows: + action = await _export(store, key, rows, dry_run=dry_run, rapport=rapport) + else: + action = "vide, rien à exporter" + + if dry_run: + print(f"{key} : {len(rows)} ligne(s), {action}, suppression simulée.") + return + + async with engine.begin() as conn: + await drop_chunk(conn, chunk) + + rapport.supprimes += 1 + print(f"{key} : {len(rows)} ligne(s), {action}, chunk {chunk.qualified_name} supprimé.") + + +async def archive_reading_chunks( + engine: AsyncEngine, + store: ArchiveStore, + *, + older_than: datetime, + dry_run: bool, +) -> Rapport: + rapport = Rapport() + + async with engine.connect() as conn: + chunks = await eligible_chunks(conn, older_than) + + rapport.chunks_vus = len(chunks) + print( + f"{len(chunks)} chunk(s) de reading entièrement antérieur(s) au {older_than.isoformat()}." + ) + + for chunk in chunks: + await _archive_chunk(engine, store, chunk, dry_run=dry_run, rapport=rapport) + + bilan = "Dry-run terminé : rien n'a été écrit ni supprimé." if dry_run else "Archivage terminé." + print( + f"{bilan} Chunks vus : {rapport.chunks_vus}, exportés : {rapport.exportes}, " + f"déjà présents : {rapport.deja_presents}, supprimés : {rapport.supprimes}, " + f"lignes : {rapport.lignes}." + ) + return rapport + + +async def _run( + settings: Settings, + store: ArchiveStore, + *, + older_than: datetime, + dry_run: bool, +) -> Rapport: + engine = create_async_engine(str(settings.database_url), pool_pre_ping=True) + try: + return await archive_reading_chunks(engine, store, older_than=older_than, dry_run=dry_run) + finally: + await engine.dispose() + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="python -m app.etl.reading_retention", + description=( + "Exporte vers Garage puis supprime les chunks de reading entièrement plus vieux " + "que la borne de rétention." + ), + ) + parser.add_argument( + "--older-than-days", + type=int, + default=None, + help="Borne en jours, par défaut APP_READING_RETENTION_DAYS.", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Liste et mesure les chunks éligibles sans rien écrire ni supprimer.", + ) + return parser + + +def main(argv: list[str] | None = None) -> None: + args = build_parser().parse_args(argv) + settings = get_settings() + + jours = ( + settings.reading_retention_days if args.older_than_days is None else args.older_than_days + ) + older_than = datetime.now(UTC) - timedelta(days=jours) + store = build_archive_store(settings) + + asyncio.run(_run(settings, store, older_than=older_than, dry_run=args.dry_run)) + + +if __name__ == "__main__": + main() diff --git a/apps/backend/pyproject.toml b/apps/backend/pyproject.toml index e884616..73666f8 100644 --- a/apps/backend/pyproject.toml +++ b/apps/backend/pyproject.toml @@ -19,6 +19,7 @@ dependencies = [ "aiosmtplib>=5.1.3", "httpx>=0.28.1", "pandas>=3.0.5", + "boto3>=1.43.101", ] [dependency-groups] @@ -29,6 +30,7 @@ dev = [ "pytest-asyncio>=1.4.0", "pytest-cov>=7.1.0", "pandas-stubs>=3.0.5.260914", + "types-boto3[s3]>=1.43.101", ] [build-system] diff --git a/apps/backend/tests/etl/test_reading_retention.py b/apps/backend/tests/etl/test_reading_retention.py new file mode 100644 index 0000000..123c3b0 --- /dev/null +++ b/apps/backend/tests/etl/test_reading_retention.py @@ -0,0 +1,702 @@ +import base64 +import io +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from datetime import UTC, datetime, timedelta, timezone +from decimal import Decimal +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import boto3 +import pandas as pd +import pytest +from botocore.exceptions import ClientError +from botocore.response import StreamingBody +from botocore.stub import Stubber +from pydantic import SecretStr +from tests.factories import make_settings + +import app.etl.reading_retention as reading_retention +from app.core.config import Settings +from app.etl.reading_retention import ( + CHUNK_ROWS, + DROP_CHUNK, + ELIGIBLE_CHUNKS, + ArchiveStore, + Chunk, + Rapport, + archive_reading_chunks, + build_archive_store, + build_parser, + decode_sse_key, + drop_chunk, + eligible_chunks, + object_key, + read_chunk_rows, + serialize_csv_gzip, + sha256_of, +) + +CLE_SSE = b"0123456789abcdef0123456789abcdef" +CLE_SSE_BASE64 = base64.b64encode(CLE_SSE).decode("ascii") + +CHUNK = Chunk( + schema="_timescaledb_internal", + name="_hyper_1_7_chunk", + range_start=datetime(2023, 1, 5, tzinfo=UTC), + range_end=datetime(2023, 1, 12, tzinfo=UTC), +) +CLE_ATTENDUE = "reading/2023/reading_20230105T000000Z_20230112T000000Z.csv.gz" + + +def make_row(**overrides: Any) -> dict[str, Any]: + ligne: dict[str, Any] = { + "reading_id": 1, + "site_id": "SITE001", + "timestamp": datetime(2023, 1, 5, 12, tzinfo=UTC), + "source": "csv", + "dataset_id": 1, + "consumption_kw": Decimal("87.34"), + "data_quality": "good", + "null_reasons": ["sensor_offline"], + "imputed_values": None, + "raw_data": {"b": 1, "a": "é"}, + } + return {**ligne, **overrides} + + +def settings_s3(**overrides: Any) -> Settings: + reglages: dict[str, Any] = { + "_env_file": None, + "secret_key": SecretStr("secret-de-test-assez-long-pour-le-validateur"), + "database_url": "postgresql+asyncpg://retention:test@localhost:5432/enervision", + "s3_endpoint_url": "http://garage:3900", + "s3_access_key": "GK0123456789", + "s3_secret_key": SecretStr("un-secret-garage"), + "s3_bucket": "enervision-archives", + "s3_sse_key": SecretStr(CLE_SSE_BASE64), + } + return Settings(**{**reglages, **overrides}) + + +def s3_client() -> Any: + return boto3.client( + "s3", + endpoint_url="http://garage:3900", + aws_access_key_id="GK0123456789", + aws_secret_access_key="un-secret-garage", + region_name="garage", + ) + + +def streaming(data: bytes) -> StreamingBody: + return StreamingBody(io.BytesIO(data), len(data)) + + +def test_settings_treat_empty_s3_values_as_absent() -> None: + settings = make_settings( + s3_endpoint_url="", s3_access_key="", s3_secret_key="", s3_bucket="", s3_sse_key="" + ) + + assert settings.s3_endpoint_url is None + assert settings.s3_access_key is None + assert settings.s3_secret_key is None + assert settings.s3_bucket is None + assert settings.s3_sse_key is None + assert settings.reading_retention_days == 1095 + + +def test_object_key_places_the_chunk_under_the_year_of_its_start() -> None: + assert object_key(CHUNK) == CLE_ATTENDUE + + +def test_object_key_expresses_the_bounds_in_utc() -> None: + paris = timezone(timedelta(hours=1)) + chunk = Chunk( + schema=CHUNK.schema, + name=CHUNK.name, + range_start=datetime(2023, 1, 5, 1, tzinfo=paris), + range_end=datetime(2023, 1, 12, 1, tzinfo=paris), + ) + + assert object_key(chunk) == CLE_ATTENDUE + + +def test_serialize_csv_gzip_is_read_back_by_pandas() -> None: + archive = serialize_csv_gzip([make_row(), make_row(reading_id=2, null_reasons=[])]) + + relu = pd.read_csv(io.BytesIO(archive), compression="gzip") + + assert list(relu.columns) == list(make_row()) + assert relu["reading_id"].tolist() == [1, 2] + assert relu["site_id"].tolist() == ["SITE001", "SITE001"] + + +def test_serialize_csv_gzip_writes_jsonb_and_arrays_as_sorted_json() -> None: + archive = serialize_csv_gzip([make_row()]) + + relu = pd.read_csv(io.BytesIO(archive), compression="gzip") + + assert relu.loc[0, "raw_data"] == '{"a": "é", "b": 1}' + assert relu.loc[0, "null_reasons"] == '["sensor_offline"]' + + +def test_serialize_csv_gzip_is_byte_for_byte_reproducible() -> None: + lignes = [make_row(), make_row(reading_id=2)] + + assert serialize_csv_gzip(lignes) == serialize_csv_gzip(lignes) + + +def test_serialize_csv_gzip_refuses_an_empty_export() -> None: + with pytest.raises(ValueError, match="Aucune ligne"): + serialize_csv_gzip([]) + + +def test_sha256_of_hashes_the_bytes() -> None: + assert sha256_of(b"hello").startswith("2cf24dba") + + +def test_store_put_sends_the_sse_c_headers_when_a_key_is_set() -> None: + client = s3_client() + store = ArchiveStore(client, bucket="enervision-archives", sse_key=CLE_SSE) + + with Stubber(client) as stub: + stub.add_response( + "put_object", + {}, + expected_params={ + "Bucket": "enervision-archives", + "Key": CLE_ATTENDUE, + "Body": b"corps", + "ContentType": "text/csv", + "ContentEncoding": "gzip", + "Metadata": {"sha256": "abc"}, + "SSECustomerAlgorithm": "AES256", + "SSECustomerKey": CLE_SSE, + }, + ) + store.put(CLE_ATTENDUE, b"corps", {"sha256": "abc"}) + stub.assert_no_pending_responses() + + +def test_store_put_omits_the_sse_c_headers_without_a_key() -> None: + client = s3_client() + store = ArchiveStore(client, bucket="enervision-archives", sse_key=None) + + with Stubber(client) as stub: + stub.add_response( + "put_object", + {}, + expected_params={ + "Bucket": "enervision-archives", + "Key": CLE_ATTENDUE, + "Body": b"corps", + "ContentType": "text/csv", + "ContentEncoding": "gzip", + "Metadata": {}, + }, + ) + store.put(CLE_ATTENDUE, b"corps", {}) + stub.assert_no_pending_responses() + + +def test_store_fetch_sha256_hashes_the_object_read_with_the_key() -> None: + client = s3_client() + store = ArchiveStore(client, bucket="enervision-archives", sse_key=CLE_SSE) + + with Stubber(client) as stub: + stub.add_response( + "get_object", + {"Body": streaming(b"hello")}, + expected_params={ + "Bucket": "enervision-archives", + "Key": CLE_ATTENDUE, + "SSECustomerAlgorithm": "AES256", + "SSECustomerKey": CLE_SSE, + }, + ) + + assert store.fetch_sha256(CLE_ATTENDUE) == sha256_of(b"hello") + + +@pytest.mark.parametrize( + ("code", "statut"), + [("NoSuchKey", 404), ("NotFound", 404), ("NoSuchKey", 400)], + ids=["no_such_key", "404_sans_code_connu", "no_such_key_sans_404"], +) +def test_store_fetch_sha256_returns_none_for_a_missing_object(code: str, statut: int) -> None: + client = s3_client() + store = ArchiveStore(client, bucket="enervision-archives", sse_key=None) + + with Stubber(client) as stub: + stub.add_client_error("get_object", service_error_code=code, http_status_code=statut) + + assert store.fetch_sha256(CLE_ATTENDUE) is None + + +def test_store_fetch_sha256_raises_any_other_error() -> None: + client = s3_client() + store = ArchiveStore(client, bucket="enervision-archives", sse_key=None) + + with Stubber(client) as stub: + stub.add_client_error("get_object", service_error_code="AccessDenied", http_status_code=403) + + with pytest.raises(ClientError): + store.fetch_sha256(CLE_ATTENDUE) + + +def test_decode_sse_key_returns_none_without_a_key() -> None: + assert decode_sse_key(None) is None + + +def test_decode_sse_key_decodes_the_base64_key() -> None: + assert decode_sse_key(SecretStr(CLE_SSE_BASE64)) == CLE_SSE + + +def test_decode_sse_key_refuses_a_key_of_the_wrong_length() -> None: + courte = base64.b64encode(b"trop-courte").decode("ascii") + + with pytest.raises(ValueError, match="exactement 32 octets"): + decode_sse_key(SecretStr(courte)) + + +@pytest.mark.parametrize( + "manquant", + ["s3_endpoint_url", "s3_access_key", "s3_secret_key", "s3_bucket"], +) +def test_build_archive_store_refuses_a_missing_setting(manquant: str) -> None: + with pytest.raises(ValueError, match="APP_S3_ENDPOINT_URL"): + build_archive_store(settings_s3(**{manquant: None})) + + +def test_build_archive_store_refuses_a_sse_key_of_the_wrong_length() -> None: + courte = base64.b64encode(b"trop-courte").decode("ascii") + + with pytest.raises(ValueError, match="exactement 32 octets"): + build_archive_store(settings_s3(s3_sse_key=SecretStr(courte))) + + +def test_build_archive_store_configures_the_client_from_the_settings( + monkeypatch: pytest.MonkeyPatch, +) -> None: + recu: dict[str, Any] = {} + + def faux_client(service: str, **kwargs: Any) -> MagicMock: + recu["service"] = service + recu.update(kwargs) + return MagicMock() + + monkeypatch.setattr(reading_retention.boto3, "client", faux_client) + + build_archive_store(settings_s3()) + + assert recu == { + "service": "s3", + "endpoint_url": "http://garage:3900", + "aws_access_key_id": "GK0123456789", + "aws_secret_access_key": "un-secret-garage", + "region_name": "garage", + } + + +def test_build_archive_store_uses_the_bucket_and_the_decoded_key() -> None: + store = build_archive_store(settings_s3()) + + with Stubber(store._client) as stub: + stub.add_response( + "get_object", + {"Body": streaming(b"hello")}, + expected_params={ + "Bucket": "enervision-archives", + "Key": CLE_ATTENDUE, + "SSECustomerAlgorithm": "AES256", + "SSECustomerKey": CLE_SSE, + }, + ) + + assert store.fetch_sha256(CLE_ATTENDUE) == sha256_of(b"hello") + + +def test_build_archive_store_accepts_an_absent_sse_key() -> None: + store = build_archive_store(settings_s3(s3_sse_key=None)) + + with Stubber(store._client) as stub: + stub.add_response( + "get_object", + {"Body": streaming(b"hello")}, + expected_params={"Bucket": "enervision-archives", "Key": CLE_ATTENDUE}, + ) + + assert store.fetch_sha256(CLE_ATTENDUE) == sha256_of(b"hello") + + +class FakeResult: + def __init__(self, rows: list[Any]) -> None: + self._rows = rows + + def mappings(self) -> FakeResult: + return self + + def scalars(self) -> FakeResult: + return self + + def all(self) -> list[Any]: + return self._rows + + +def chunk_mapping(chunk: Chunk) -> dict[str, Any]: + return { + "chunk_schema": chunk.schema, + "chunk_name": chunk.name, + "range_start": chunk.range_start, + "range_end": chunk.range_end, + } + + +async def test_eligible_chunks_queries_the_timescaledb_catalog() -> None: + conn = AsyncMock() + conn.execute.return_value = FakeResult([chunk_mapping(CHUNK)]) + borne = datetime(2023, 10, 1, tzinfo=UTC) + + chunks = await eligible_chunks(conn, borne) + + assert chunks == [CHUNK] + statement, params = conn.execute.await_args.args + assert statement is ELIGIBLE_CHUNKS + assert params == {"older_than": borne} + + +async def test_read_chunk_rows_reads_through_the_hypertable_within_the_chunk_bounds() -> None: + conn = AsyncMock() + conn.execute.return_value = FakeResult([make_row(), make_row(reading_id=2)]) + + lignes = await read_chunk_rows(conn, CHUNK) + + assert lignes == [make_row(), make_row(reading_id=2)] + statement, params = conn.execute.await_args.args + assert statement is CHUNK_ROWS + assert params == {"start": CHUNK.range_start, "end": CHUNK.range_end} + + +async def test_drop_chunk_targets_the_chunk_by_its_own_bounds() -> None: + conn = AsyncMock() + conn.execute.return_value = FakeResult([CHUNK.qualified_name]) + + await drop_chunk(conn, CHUNK) + + statement, params = conn.execute.await_args.args + assert statement is DROP_CHUNK + assert params == {"start": CHUNK.range_start, "end": CHUNK.range_end} + + +@pytest.mark.parametrize( + "rendu", + [[], ["_timescaledb_internal._hyper_1_7_chunk", "_timescaledb_internal._hyper_1_8_chunk"]], + ids=["aucun_chunk", "deux_chunks"], +) +async def test_drop_chunk_raises_unless_exactly_the_chunk_was_dropped(rendu: list[str]) -> None: + conn = AsyncMock() + conn.execute.return_value = FakeResult(rendu) + + with pytest.raises(RuntimeError, match=r"exactement _timescaledb_internal\._hyper_1_7_chunk"): + await drop_chunk(conn, CHUNK) + + +class FakeConn: + def __init__(self, journal: list[str], chunks: list[Chunk], rows: list[dict[str, Any]]) -> None: + self._journal = journal + self._chunks = chunks + self._rows = rows + + async def execute(self, statement: Any, params: dict[str, Any]) -> FakeResult: + if statement is ELIGIBLE_CHUNKS: + self._journal.append("lister") + return FakeResult([chunk_mapping(chunk) for chunk in self._chunks]) + if statement is CHUNK_ROWS: + self._journal.append("lire") + return FakeResult(self._rows) + self._journal.append("drop") + chunk = next(c for c in self._chunks if c.range_start == params["start"]) + return FakeResult([chunk.qualified_name]) + + +class FakeEngine: + def __init__(self, conn: FakeConn, journal: list[str]) -> None: + self._conn = conn + self._journal = journal + + @asynccontextmanager + async def connect(self) -> AsyncIterator[FakeConn]: + self._journal.append("connect") + yield self._conn + + @asynccontextmanager + async def begin(self) -> AsyncIterator[FakeConn]: + self._journal.append("begin") + yield self._conn + + async def dispose(self) -> None: + self._journal.append("dispose") + + +class FakeStore(ArchiveStore): + def __init__(self, journal: list[str], *, corrompt: bool = False) -> None: + super().__init__(MagicMock(), bucket="enervision-archives", sse_key=None) + self._journal = journal + self._corrompt = corrompt + self.objets: dict[str, str] = {} + self.metadata: dict[str, dict[str, str]] = {} + + def put(self, key: str, body: bytes, metadata: dict[str, str]) -> None: + self._journal.append("put") + self.objets[key] = "sha-corrompu" if self._corrompt else sha256_of(body) + self.metadata[key] = metadata + + def fetch_sha256(self, key: str) -> str | None: + self._journal.append("relire") + return self.objets.get(key) + + +def make_archive( + chunks: list[Chunk] | None = None, + rows: list[dict[str, Any]] | None = None, + *, + corrompt: bool = False, +) -> tuple[FakeEngine, FakeStore, list[str]]: + journal: list[str] = [] + lignes = [make_row(), make_row(reading_id=2)] if rows is None else rows + eligibles = [CHUNK] if chunks is None else chunks + engine = FakeEngine(FakeConn(journal, eligibles, lignes), journal) + return engine, FakeStore(journal, corrompt=corrompt), journal + + +async def test_archive_reading_chunks_reads_exports_verifies_then_drops( + capsys: pytest.CaptureFixture[str], +) -> None: + engine, store, journal = make_archive() + borne = datetime(2023, 10, 1, tzinfo=UTC) + + rapport = await archive_reading_chunks(engine, store, older_than=borne, dry_run=False) + + assert journal == [ + "connect", + "lister", + "connect", + "lire", + "relire", + "put", + "relire", + "begin", + "drop", + ] + assert rapport == Rapport(chunks_vus=1, exportes=1, deja_presents=0, supprimes=1, lignes=2) + assert store.objets[CLE_ATTENDUE] == sha256_of( + serialize_csv_gzip([make_row(), make_row(reading_id=2)]) + ) + assert store.metadata[CLE_ATTENDUE] == {"sha256": store.objets[CLE_ATTENDUE], "rows": "2"} + + sortie = capsys.readouterr().out + assert f"{CLE_ATTENDUE} : 2 ligne(s)" in sortie + assert "exportés et relus" in sortie + assert f"chunk {CHUNK.qualified_name} supprimé" in sortie + assert "Archivage terminé." in sortie + + +async def test_archive_reading_chunks_skips_the_upload_when_the_object_already_matches() -> None: + engine, store, journal = make_archive() + store.objets[CLE_ATTENDUE] = sha256_of(serialize_csv_gzip([make_row(), make_row(reading_id=2)])) + + rapport = await archive_reading_chunks( + engine, store, older_than=datetime(2023, 10, 1, tzinfo=UTC), dry_run=False + ) + + assert "put" not in journal + assert journal[-2:] == ["begin", "drop"] + assert rapport == Rapport(chunks_vus=1, exportes=0, deja_presents=1, supprimes=1, lignes=2) + + +async def test_archive_reading_chunks_re_uploads_when_the_stored_object_differs() -> None: + engine, store, journal = make_archive() + store.objets[CLE_ATTENDUE] = "un-autre-sha" + + rapport = await archive_reading_chunks( + engine, store, older_than=datetime(2023, 10, 1, tzinfo=UTC), dry_run=False + ) + + assert journal.count("put") == 1 + assert rapport.exportes == 1 + assert rapport.deja_presents == 0 + + +async def test_archive_reading_chunks_in_dry_run_neither_writes_nor_drops( + capsys: pytest.CaptureFixture[str], +) -> None: + engine, store, journal = make_archive() + + rapport = await archive_reading_chunks( + engine, store, older_than=datetime(2023, 10, 1, tzinfo=UTC), dry_run=True + ) + + assert "put" not in journal + assert "begin" not in journal + assert "drop" not in journal + assert rapport == Rapport(chunks_vus=1, exportes=0, deja_presents=0, supprimes=0, lignes=2) + + sortie = capsys.readouterr().out + assert "octets à exporter, suppression simulée." in sortie + assert "Dry-run terminé : rien n'a été écrit ni supprimé." in sortie + + +async def test_archive_reading_chunks_keeps_the_chunk_when_the_read_back_differs() -> None: + engine, store, journal = make_archive(corrompt=True) + + with pytest.raises(RuntimeError, match="sha256 sha-corrompu au lieu de"): + await archive_reading_chunks( + engine, store, older_than=datetime(2023, 10, 1, tzinfo=UTC), dry_run=False + ) + + assert "put" in journal + assert "drop" not in journal + + +async def test_archive_reading_chunks_drops_an_empty_chunk_without_exporting( + capsys: pytest.CaptureFixture[str], +) -> None: + engine, store, journal = make_archive(rows=[]) + + rapport = await archive_reading_chunks( + engine, store, older_than=datetime(2023, 10, 1, tzinfo=UTC), dry_run=False + ) + + assert "put" not in journal + assert "relire" not in journal + assert journal[-2:] == ["begin", "drop"] + assert rapport == Rapport(chunks_vus=1, exportes=0, deja_presents=0, supprimes=1, lignes=0) + assert "vide, rien à exporter" in capsys.readouterr().out + + +async def test_archive_reading_chunks_handles_each_chunk_in_turn() -> None: + suivant = Chunk( + schema=CHUNK.schema, + name="_hyper_1_8_chunk", + range_start=CHUNK.range_end, + range_end=CHUNK.range_end + timedelta(days=7), + ) + engine, store, journal = make_archive(chunks=[CHUNK, suivant]) + + rapport = await archive_reading_chunks( + engine, store, older_than=datetime(2023, 10, 1, tzinfo=UTC), dry_run=False + ) + + assert rapport == Rapport(chunks_vus=2, exportes=2, deja_presents=0, supprimes=2, lignes=4) + assert set(store.objets) == {CLE_ATTENDUE, object_key(suivant)} + assert journal.count("drop") == 2 + + +async def test_archive_reading_chunks_reports_nothing_to_do_without_eligible_chunks( + capsys: pytest.CaptureFixture[str], +) -> None: + engine, store, journal = make_archive(chunks=[]) + + rapport = await archive_reading_chunks( + engine, store, older_than=datetime(2023, 10, 1, tzinfo=UTC), dry_run=False + ) + + assert rapport == Rapport() + assert journal == ["connect", "lister"] + assert "0 chunk(s) de reading" in capsys.readouterr().out + + +def test_build_parser_defaults_to_the_settings_and_a_real_run() -> None: + arguments = build_parser().parse_args([]) + + assert arguments.older_than_days is None + assert arguments.dry_run is False + + +def test_build_parser_reads_the_bound_and_the_dry_run() -> None: + arguments = build_parser().parse_args(["--older-than-days", "400", "--dry-run"]) + + assert arguments.older_than_days == 400 + assert arguments.dry_run is True + + +def test_build_parser_refuses_a_non_integer_bound() -> None: + with pytest.raises(SystemExit): + build_parser().parse_args(["--older-than-days", "un-an"]) + + +def test_build_parser_answers_help_without_settings(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("APP_SECRET_KEY", raising=False) + + with pytest.raises(SystemExit) as sortie: + build_parser().parse_args(["--help"]) + + assert sortie.value.code == 0 + + +@pytest.fixture +def main_branche(monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]: + capture: dict[str, Any] = {} + journal: list[str] = [] + engine = FakeEngine(FakeConn(journal, [], []), journal) + store = FakeStore(journal) + + async def faux_archive( + engine_recu: Any, store_recu: Any, *, older_than: datetime, dry_run: bool + ) -> Rapport: + capture.update(engine=engine_recu, store=store_recu, older_than=older_than, dry_run=dry_run) + return Rapport() + + def faux_engine(url: str, **kwargs: Any) -> FakeEngine: + capture["url"] = url + capture["engine_kwargs"] = kwargs + return engine + + monkeypatch.setattr(reading_retention, "get_settings", settings_s3) + monkeypatch.setattr(reading_retention, "build_archive_store", lambda settings: store) + monkeypatch.setattr(reading_retention, "create_async_engine", faux_engine) + monkeypatch.setattr(reading_retention, "archive_reading_chunks", faux_archive) + capture["journal"] = journal + capture["store_attendu"] = store + capture["engine_attendu"] = engine + return capture + + +def test_main_uses_the_retention_setting_by_default(main_branche: dict[str, Any]) -> None: + avant = datetime.now(UTC) + + reading_retention.main([]) + + attendu = avant - timedelta(days=1095) + assert timedelta(0) <= main_branche["older_than"] - attendu < timedelta(seconds=5) + assert main_branche["dry_run"] is False + assert main_branche["store"] is main_branche["store_attendu"] + assert main_branche["engine"] is main_branche["engine_attendu"] + assert main_branche["url"] == "postgresql+asyncpg://retention:test@localhost:5432/enervision" + assert main_branche["engine_kwargs"] == {"pool_pre_ping": True} + assert main_branche["journal"] == ["dispose"] + + +def test_main_honours_an_explicit_bound_and_the_dry_run(main_branche: dict[str, Any]) -> None: + avant = datetime.now(UTC) + + reading_retention.main(["--older-than-days", "10", "--dry-run"]) + + attendu = avant - timedelta(days=10) + assert timedelta(0) <= main_branche["older_than"] - attendu < timedelta(seconds=5) + assert main_branche["dry_run"] is True + + +def test_main_fails_before_touching_the_database_without_s3_settings( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(reading_retention, "get_settings", lambda: settings_s3(s3_bucket=None)) + monkeypatch.setattr( + reading_retention, + "create_async_engine", + lambda *_, **__: pytest.fail("l'engine ne doit pas être créé"), + ) + + with pytest.raises(ValueError, match="APP_S3_BUCKET"): + reading_retention.main([]) diff --git a/apps/backend/uv.lock b/apps/backend/uv.lock index 7b67e36..2ec1d36 100644 --- a/apps/backend/uv.lock +++ b/apps/backend/uv.lock @@ -192,6 +192,43 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/3c/d7/8fb3044eaef08a310acfe23dae9a8e2e07d305edc29a53497e52bc76eca7/asyncpg-0.31.0-cp314-cp314t-win_amd64.whl", hash = "sha256:bd4107bb7cdd0e9e65fae66a62afd3a249663b844fa34d479f6d5b3bef9c04c3", size = 706062, upload-time = "2025-11-24T23:26:44.086Z" }, ] +[[package]] +name = "boto3" +version = "1.43.101" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "botocore" }, + { name = "jmespath" }, + { name = "s3transfer" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/ad/ef/096f1520a4b0cbc794348fcf77ada637e5f98145c3453f219d678c3a0798/boto3-1.43.101.tar.gz", hash = "sha256:49f3eb750f70e050df9929a7e9392e67896c97d7d0a448f13ed3354c634268bd", size = 112635, upload-time = "2026-09-23T19:23:29.553Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d7/73/8dd65374f88b1b2a33656d9c808dc36aef0cb4a22c74d618ba6fe0092cd2/boto3-1.43.101-py3-none-any.whl", hash = "sha256:8a899b0ea94df3f2fab6d0c69caf2791f2971449696834374d8e88ced01c7ef3", size = 140041, upload-time = "2026-09-23T19:23:27.61Z" }, +] + +[[package]] +name = "botocore" +version = "1.43.101" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "jmespath" }, + { name = "python-dateutil" }, + { name = "urllib3" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/12/12/e90cc51bd65ecdcd0eedcd522d3c9f102b1d2c601f39f1f1c256695d63a3/botocore-1.43.101.tar.gz", hash = "sha256:3bc67fb55046e1e05ce5f2bd0171f37bef1cf54161786ef04ff338614d98169e", size = 16202504, upload-time = "2026-09-23T19:23:24.49Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/dd/0d/6679253333d6ba74b8ad7077560687096629255ef526e106bb3accceffcc/botocore-1.43.101-py3-none-any.whl", hash = "sha256:f380237ffecc3f887265cd09c4d7e9c8e8dd9ba6162af83b1fc9e5d24622e461", size = 15897867, upload-time = "2026-09-23T19:23:21.643Z" }, +] + +[[package]] +name = "botocore-stubs" +version = "1.43.67" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/3f/45/53d662227dc4787b2c854445ee7eb4751cb5d74cfb5c686a6ecbe1f94c17/botocore_stubs-1.43.67.tar.gz", hash = "sha256:853e74014a1f557055c4ffae5fb38d7c65c7c0520e1aab366cac41d5428f419d", size = 42846, upload-time = "2026-08-08T14:57:53.412Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/4e/5e/bdbf19967898a032292da65a47d6e25b2eee55865db4e687f861d80b5602/botocore_stubs-1.43.67-py3-none-any.whl", hash = "sha256:c51262bac3341c1cda71f05fa01141fffd3990d7a92c7960e3b755c1bc830373", size = 67244, upload-time = "2026-08-08T14:57:52.01Z" }, +] + [[package]] name = "certifi" version = "2026.7.22" @@ -325,6 +362,7 @@ dependencies = [ { name = "anyio" }, { name = "argon2-cffi" }, { name = "asyncpg" }, + { name = "boto3" }, { name = "fastapi" }, { name = "httpx" }, { name = "pandas" }, @@ -345,6 +383,7 @@ dev = [ { name = "pytest-asyncio" }, { name = "pytest-cov" }, { name = "ruff" }, + { name = "types-boto3", extra = ["s3"] }, ] [package.metadata] @@ -354,6 +393,7 @@ requires-dist = [ { name = "anyio", specifier = ">=4.0" }, { name = "argon2-cffi", specifier = ">=23.1" }, { name = "asyncpg", specifier = ">=0.31.0" }, + { name = "boto3", specifier = ">=1.43.101" }, { name = "fastapi", specifier = ">=0.141.1" }, { name = "httpx", specifier = ">=0.28.1" }, { name = "pandas", specifier = ">=3.0.5" }, @@ -374,6 +414,7 @@ dev = [ { name = "pytest-asyncio", specifier = ">=1.4.0" }, { name = "pytest-cov", specifier = ">=7.1.0" }, { name = "ruff", specifier = ">=0.16.7" }, + { name = "types-boto3", extras = ["s3"], specifier = ">=1.43.101" }, ] [[package]] @@ -496,6 +537,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" }, ] +[[package]] +name = "jmespath" +version = "1.1.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d3/59/322338183ecda247fb5d1763a6cbe46eff7222eaeebafd9fa65d4bf5cb11/jmespath-1.1.0.tar.gz", hash = "sha256:472c87d80f36026ae83c6ddd0f1d05d4e510134ed462851fd5f754c8c3cbb88d", size = 27377, upload-time = "2026-01-22T16:35:26.279Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/14/2f/967ba146e6d58cf6a652da73885f52fc68001525b4197effc174321d70b4/jmespath-1.1.0-py3-none-any.whl", hash = "sha256:a5663118de4908c91729bea0acadca56526eb2698e83de10cd116ae0f4e97c64", size = 20419, upload-time = "2026-01-22T16:35:24.919Z" }, +] + [[package]] name = "librt" version = "0.15.0" @@ -959,6 +1009,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/fe/a0/50787329e4f20bf9dc9f6230015d46ec69c51a97ace5bc202dae4755365d/ruff-0.16.8-py3-none-win_arm64.whl", hash = "sha256:d075e820af612102ce217f07cc93e69f9490b10ec13ea85fa87bd03d996cef8a", size = 10386316, upload-time = "2026-09-16T15:54:43.332Z" }, ] +[[package]] +name = "s3transfer" +version = "0.19.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "botocore" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/76/43/35e4d8aa320bffe8287fe8f65f578fa2d2db0a64212f0e710dce58267854/s3transfer-0.19.2.tar.gz", hash = "sha256:ba0309fd86be3c27dbf78cdd813c13c5e1df16e5874b99d2535ebbdfb9892993", size = 165592, upload-time = "2026-07-22T19:30:44.432Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/bc/e7/5c595c75e9f41a44f30e526eda465ea0b4eec93470e074e4a111b253f13a/s3transfer-0.19.2-py3-none-any.whl", hash = "sha256:d8168eccca828cbb2cd573675333f3bddd254313a9c42494b84c76b539e8ba25", size = 90216, upload-time = "2026-07-22T19:30:43.251Z" }, +] + [[package]] name = "six" version = "1.17.0" @@ -1006,6 +1068,42 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/c8/cb/6a6a47d5b464bd08695d254f3da6e7986cc70c9fa5d778eda57538edfe56/starlette-1.6.0-py3-none-any.whl", hash = "sha256:a86dd39d14bb45f85a3d18525215a9ef0cfd1f192ac793220e72598c90335f0c", size = 75969, upload-time = "2026-08-08T18:27:56.196Z" }, ] +[[package]] +name = "types-boto3" +version = "1.43.101" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "botocore-stubs" }, + { name = "types-s3transfer" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/1d/32/e9cfa9a44874cc603220713084bd3d347ee8d3f539748673aa7a62cc7b9c/types_boto3-1.43.101.tar.gz", hash = "sha256:a892e195f6b46e73a3278b08f45dea6470306662647ccf212e8e4366e052e863", size = 105304, upload-time = "2026-09-23T20:24:43.981Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/04/9e/57626d063d66db4329a6ee96524cf768a20b60194120a0d71ee962758d2b/types_boto3-1.43.101-py3-none-any.whl", hash = "sha256:a5303a8024fa0588dad70fb7ba5845adaa4d86c0ea395063d7ae33badbc6396f", size = 71672, upload-time = "2026-09-23T20:24:39.849Z" }, +] + +[package.optional-dependencies] +s3 = [ + { name = "types-boto3-s3" }, +] + +[[package]] +name = "types-boto3-s3" +version = "1.43.93" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f7/1a/285aa2a27436e437aea1c6d6f964b692df3d8c349bf6a35fd476d0b2f7bb/types_boto3_s3-1.43.93.tar.gz", hash = "sha256:6a7f979872b81f6bf22eb4dc39ea9909d635ec756275eca69e9caabdc94d5a6a", size = 79218, upload-time = "2026-09-11T19:44:38.149Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/1f/16/db644e738b967336fb0ca335d708c7d659a965b8ae703e9c50fe209c59be/types_boto3_s3-1.43.93-py3-none-any.whl", hash = "sha256:da9249f05ea081bb3b3f3b8cc49099a988ff5c89da8a7393532397c6174e04d3", size = 86538, upload-time = "2026-09-11T19:44:36.68Z" }, +] + +[[package]] +name = "types-s3transfer" +version = "0.16.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/fe/64/42689150509eb3e6e82b33ee3d89045de1592488842ddf23c56957786d05/types_s3transfer-0.16.0.tar.gz", hash = "sha256:b4636472024c5e2b62278c5b759661efeb52a81851cde5f092f24100b1ecb443", size = 13557, upload-time = "2025-12-08T08:13:09.928Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/98/27/e88220fe6274eccd3bdf95d9382918716d312f6f6cef6a46332d1ee2feff/types_s3transfer-0.16.0-py3-none-any.whl", hash = "sha256:1c0cd111ecf6e21437cb410f5cddb631bfb2263b77ad973e79b9c6d0cb24e0ef", size = 19247, upload-time = "2025-12-08T08:13:08.426Z" }, +] + [[package]] name = "typing-extensions" version = "4.16.0" @@ -1036,6 +1134,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/f9/bc/8737e8d54cf51106118039b83f485a4783112fab49ea9d044b234978a46e/tzdata-2026.4-py2.py3-none-any.whl", hash = "sha256:c2169a8b0a7a5e9674da5a135ccdfb2b3e671b333ed9fed17b41f73c34476e81", size = 347494, upload-time = "2026-09-12T12:56:01.67Z" }, ] +[[package]] +name = "urllib3" +version = "2.8.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/e3/05/b17359e1cefb4f909b5e40b1b90a496d987258916dbbf88e842c729f510e/urllib3-2.8.0.tar.gz", hash = "sha256:63bf2ead4c879426ebf22ef2a781eeb4aa3b4ae798a0435506f8687fd5bb9b63", size = 458972, upload-time = "2026-09-15T19:29:36.253Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/92/9d/c4e665119135114480843e7ab388fa94d8480650450e6f8e26b70d323a4c/urllib3-2.8.0-py3-none-any.whl", hash = "sha256:0cf3cae568d36aa9576b28dfb35f11328f1cb974ca7647d9475ebb86c75ac6e3", size = 135717, upload-time = "2026-09-15T19:29:34.577Z" }, +] + [[package]] name = "uvicorn" version = "0.53.0" diff --git a/docs/architecture/20-backend.md b/docs/architecture/20-backend.md index 4e4776a..fa0dd4d 100644 --- a/docs/architecture/20-backend.md +++ b/docs/architecture/20-backend.md @@ -104,6 +104,16 @@ démarre ne prouve rien sur la base, la première connexion réelle a lieu au pr | `APP_TRUST_PROXY_HEADERS` | `false` | À vrai derrière un proxy, sinon le compteur par IP devient global | | `APP_EXPOSE_API_DOCS` | déduit | Faux en `staging` et `prod` si non renseigné | | `APP_METRICS_TOKEN` | absent | Si présent et non vide, `/metrics` exige `Authorization: Bearer`. Vide vaut absent | +| `APP_S3_ENDPOINT_URL` | absent | Endpoint S3 des archives ; `http://garage:3900` posé par Compose sur `airflow-scheduler`. Vide vaut absent | +| `APP_S3_REGION` | `garage` | Région déclarée au client S3 | +| `APP_S3_ACCESS_KEY` | absent | Identifiant de la clé Garage. Vide vaut absent | +| `APP_S3_SECRET_KEY` | absent | Secret de la clé Garage, `SecretStr`. Vide vaut absent | +| `APP_S3_BUCKET` | absent | Bucket des archives, `enervision-archives` en Compose. Vide vaut absent | +| `APP_S3_SSE_KEY` | absent | Base64 de 32 octets, clé SSE-C des archives, `SecretStr`. Vide vaut absent | +| `APP_READING_RETENTION_DAYS` | `1095` | Profondeur de `reading` en base chaude, 30 jours minimum | + +L'API n'exige aucun des réglages `APP_S3_*` ni `APP_READING_RETENTION_DAYS` : seul +`app.etl.reading_retention` les réclame, et refuse de partir sans endpoint, clés et bucket. Cinq gardes refusent de démarrer plutôt que de laisser passer une erreur silencieuse : secret de moins de 32 caractères ou laissé à sa valeur d'exemple, `debug` en `staging` ou diff --git a/docs/architecture/40-data.md b/docs/architecture/40-data.md index 91ae347..81f5387 100644 --- a/docs/architecture/40-data.md +++ b/docs/architecture/40-data.md @@ -16,8 +16,9 @@ L'ingestion des **mesures** est implémentée pour les deux sources du MVP, le d l'API Mock. Celle des **alertes** de l'API Mock, `/alerts`, reste à faire : voir l'[ADR 0006](../adr/0006-moteur-de-regles-dans-le-backend.md). Les alertes `source='enervision'`, elles, sont produites par la détection interne, désormais ordonnancée par le DAG Airflow `alertes` -(issue #116). L'orchestration de l'ingestion, les agrégats continus, la compression et la -rétention restent des cibles. +(issue #116). L'orchestration de l'ingestion, les agrégats continus et la compression restent +des cibles. La rétention de `reading` est faite : chaque chunk plus vieux que la borne est exporté +vers Garage puis supprimé (issue #36, section « Rétention et archivage » ci-dessous). ## Trois emplacements, trois rôles @@ -27,7 +28,7 @@ au mauvais endroit ne s'exécute jamais, ou s'exécute deux fois. | Emplacement | Contenu | Quand ça s'exécute | |---|---|---| | `db/init/` | Extensions, bases annexes | **Une seule fois**, à la première initialisation du conteneur, quand `PGDATA` est vide. Ne rejoue jamais | -| `db/migrations/` | SQL versionné qui ne découle pas du schéma applicatif : rétention, compression | À la main, aujourd'hui vide | +| `db/migrations/` | SQL versionné qui ne découle pas du schéma applicatif : compression. La rétention de `reading` n'y est pas : une politique TimescaleDB ignorerait l'export, elle vit dans `apps/backend/app/etl/reading_retention.py`, ordonnancée par le DAG `retention` ([ADR 0019](../adr/0019-stockage-objet-garage-et-cycle-de-vie-des-mesures.md)) | À la main, aujourd'hui vide | | `apps/backend/alembic/` | Le schéma exposé par l'API, et lui seul | `alembic upgrade head`, c'est `Base.metadata` qui fait foi | Une hypertable relève des deux derniers : **Alembic crée la table, et le `create_hypertable()` @@ -75,8 +76,9 @@ Les mécanismes d'ingestion sont maintenant implémentés pour les deux sources Les traitements sont actuellement exécutables directement depuis le backend. -L'orchestration avec Apache Airflow reste une cible, tout comme les agrégats continus, -la compression et les politiques de rétention. +L'orchestration avec Apache Airflow reste une cible, tout comme les agrégats continus et la +compression. La rétention est faite : le DAG `retention` exporte chaque chunk de `reading` plus +vieux que `READING_RETENTION_DAYS` vers Garage, puis le supprime. ```mermaid flowchart LR @@ -91,7 +93,8 @@ flowchart LR hy -.-> agg[("Agrégat continu")] hy -.-> comp["Compression"] - hy -.-> ret["Rétention"] + hy --> ret["Rétention : export CSV gzip vers Garage, puis drop_chunks"] + ret --> garage[("Garage S3")] agg -.-> backend["API FastAPI"] agg -.-> graf["Grafana"] @@ -104,6 +107,26 @@ Les flèches pointillées représentent les éléments encore prévus comme cibl Les lectures de l'API et de Grafana viseront l'agrégat continu, pas la table brute : c'est tout l'intérêt de TimescaleDB, et cela doit rester vrai quand les volumes augmenteront. +### Rétention et archivage (issue #36) + +Statut : `Fait`. + +`apps/backend/app/etl/reading_retention.py`, ordonnancé chaque nuit à 03:20 UTC par le DAG +`retention`, sélectionne dans `timescaledb_information.chunks` les chunks de `reading` dont +`range_end` est antérieur ou égal à `now() - READING_RETENTION_DAYS` : seul un chunk entièrement +plus vieux que la borne est éligible. Chaque chunk est lu via l'hypertable (`WHERE timestamp >= +range_start AND timestamp < range_end`), jamais via la table interne, sérialisé en CSV gzip +reproductible (jsonb et tableaux en JSON trié), puis écrit chiffré SSE-C sous la clé +`reading//reading__.csv.gz`, bornes UTC compactes. L'objet est relu et son +sha256 comparé à celui du corps envoyé ; en cas d'écart le chunk est conservé. Seulement alors +`drop_chunks('reading', older_than => range_end, newer_than => range_start)` supprime ce chunk et +lui seul, dans une transaction dédiée et courte : `drop_chunks` pose un verrou exclusif sur +`reading`, `site` et `dataset` jusqu'au COMMIT. Un objet déjà présent avec le même sha256 n'est pas +réécrit et un chunk supprimé n'est plus éligible : rejouer le DAG est sans effet, `--dry-run` liste +et mesure sans rien écrire. Le premier passage en production archive les chunks de janvier à +septembre 2023 ; la démo, ancrée au 31/12/2024, n'est pas touchée. Restauration manuelle : +télécharger l'objet avec la clé SSE-C, `gunzip`, `COPY` dans `reading` ; aucune commande fournie. + ## Tables d'authentification Statut : `Fait`. @@ -239,8 +262,9 @@ colonne de temps : les index déclarés dans la révision le couvrent déjà. devient ininterprétable dès le premier changement d'heure. - **La colonne de partitionnement entre dans la clé primaire.** Dans `reading` elle s'appelle `timestamp` : c'est un nom de colonne, son type reste `timestamptz`. -- **Les politiques de rétention et de compression** vont dans `db/migrations/`, pas dans Alembic : - elles ne découlent pas du schéma applicatif. +- **Les politiques de compression** vont dans `db/migrations/`, pas dans Alembic : elles ne + découlent pas du schéma applicatif. La rétention de `reading` est un traitement ETL + (`reading_retention.py`), pas une politique TimescaleDB : elle doit exporter avant de supprimer. - **Tout modèle doit être importé dans `app/models/__init__.py`**, sans quoi `alembic revision --autogenerate` ne le voit pas et génère un `drop` de sa table. @@ -251,7 +275,9 @@ livrés : ce qui suit porte sur leur exploitation, plus sur leur forme. - **Quelle granularité** conserver à long terme à l'ingestion : seconde, minute ou quart d'heure. - **Quels agrégats continus** créer et sur quelles fenêtres. -- **Quelle profondeur de rétention** conserver en données brutes et à partir de quand compresser. +- **Quelle profondeur de rétention** : répondu par l'issue #36. Trois ans en base chaude par + défaut (`READING_RETENTION_DAYS`, 1095 jours) ; au-delà, les chunks sont archivés en CSV gzip + sur Garage, chiffrés SSE-C, puis supprimés. Reste ouvert : à partir de quand compresser. - **Multi-tenant ou non** : un site appartient-il à un client et faut-il cloisonner les lectures. ## Modélisation détaillée des données diff --git a/etl/airflow/dags/retention.py b/etl/airflow/dags/retention.py new file mode 100644 index 0000000..c5ded6e --- /dev/null +++ b/etl/airflow/dags/retention.py @@ -0,0 +1,45 @@ +"""DAG de rétention de l'hypertable `reading` : export vers Garage puis suppression (issue #36). + +La nuit, parce que `drop_chunks` pose un verrou exclusif sur `reading`, `site` et `dataset` +jusqu'au COMMIT : un chunk est supprimé dans une transaction courte, mais hors des heures où +l'API et les DAGs horaires écrivent. À :20 pour se glisser entre `ml_score` (à l'heure pile), +`alertes` (à :15) et `mock_api_import` (à :45), bien avant `derive` (05:30). Une reprise est sans +risque : le module est idempotent, un objet déjà exporté avec le même sha256 n'est pas réécrit et +un chunk déjà supprimé n'est plus éligible. La borne vient de `APP_READING_RETENTION_DAYS` +(1095 jours), posée par le compose sur `airflow-scheduler` avec les réglages `APP_S3_*`. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta + +from airflow.providers.standard.operators.bash import BashOperator +from airflow.sdk import DAG + +# Le backend a son propre environnement uv dans l'image (ADR 0008). `--no-sync` et +# `env -u VIRTUAL_ENV` : cf. `ml_train.py`, même raisonnement. +COMMANDE_BACKEND = "cd /opt/backend && env -u VIRTUAL_ENV uv run --no-sync python -m" + +TENTATIVES = 1 +DELAI_ENTRE_TENTATIVES = timedelta(minutes=5) +PLAFOND = timedelta(minutes=20) + +with DAG( + dag_id="retention", + description=( + "Exporte vers Garage puis supprime les chunks de reading plus vieux que la borne de " + "rétention (app.etl.reading_retention)." + ), + schedule="20 3 * * *", + start_date=datetime(2026, 1, 1), + catchup=False, + max_active_runs=1, + tags=["etl", "retention"], +) as dag: + BashOperator( + task_id="archiver", + bash_command=f"{COMMANDE_BACKEND} app.etl.reading_retention", + retries=TENTATIVES, + retry_delay=DELAI_ENTRE_TENTATIVES, + execution_timeout=PLAFOND, + ) diff --git a/etl/airflow/tests/test_dags.py b/etl/airflow/tests/test_dags.py index 9db4292..bdbbf5e 100644 --- a/etl/airflow/tests/test_dags.py +++ b/etl/airflow/tests/test_dags.py @@ -18,6 +18,7 @@ DAG_IDS = [ "historical_import", "mock_api_import", "derive", + "retention", ] TACHES = [ ("ml_train", "train"), @@ -27,6 +28,7 @@ TACHES = [ ("historical_import", "import_historical"), ("mock_api_import", "import_mock_api"), ("derive", "derive"), + ("retention", "archiver"), ] @@ -228,6 +230,24 @@ def test_derive_never_retries_a_detected_drift(dagbag: DagBag) -> None: assert dagbag.dags["derive"].get_task("derive").retries == 0 +def test_retention_runs_nightly(dagbag: DagBag) -> None: + # Entre `ml_score` (:00), `alertes` (:15) et `mock_api_import` (:45) : drop_chunks verrouille + # reading, site et dataset jusqu'au COMMIT. + assert dagbag.dags["retention"].timetable.expression == "20 3 * * *" + + +def test_retention_calls_the_backend_retention_module(dagbag: DagBag) -> None: + commande = dagbag.dags["retention"].get_task("archiver").bash_command + + assert "app.etl.reading_retention" in commande + + +def test_retention_runs_in_the_backend_environment(dagbag: DagBag) -> None: + commande = dagbag.dags["retention"].get_task("archiver").bash_command + + assert "/opt/backend" in commande + + @pytest.mark.parametrize(("dag_id", "task_id"), TACHES) def test_tasks_never_resync_the_baked_environment( dagbag: DagBag, dag_id: str, task_id: str