From 091ee344d1b0f012bef6104e6522748143ab6476 Mon Sep 17 00:00:00 2001
From: Johan LEROY <johan@beehelp.fr>
Date: Tue, 22 Sep 2026 09:05:53 +0200
Subject: [PATCH] refactor(etl): aligne la DAG historical_import sur le SDK
 Airflow 3
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Même mouvement que pour ml_train, ml_score et alertes (#135) : DAG depuis
airflow.sdk, BashOperator depuis le provider standard, et la planification
lue sur dag.schedule puisque les timetables n'exposent plus summary. Les
anciens imports fonctionnaient encore, avec un avertissement de dépréciation.
---
 etl/airflow/dags/historical_import.py | 90 +++++++++++++--------------
 etl/airflow/tests/test_dags.py        |  2 +-
 2 files changed, 46 insertions(+), 46 deletions(-)

diff --git a/etl/airflow/dags/historical_import.py b/etl/airflow/dags/historical_import.py
index a668578..30d1a15 100644
--- a/etl/airflow/dags/historical_import.py
+++ b/etl/airflow/dags/historical_import.py
@@ -1,45 +1,45 @@
-"""DAG d'import du dataset historique EnerVision (issue #119).
-
-Orchestre le pipeline existant `app.etl.historical_import` sans dupliquer sa logique ETL.
-Le dataset historique sert à initialiser l'environnement : le DAG reste donc manuel.
-
-Le backend est exécuté dans l'environnement `/opt/backend` embarqué dans l'image Airflow,
-sur le même patron que le DAG `alertes` (ADR 0008).
-"""
-
-from __future__ import annotations
-
-from datetime import datetime, timedelta
-
-from airflow.models.dag import DAG
-from airflow.operators.bash import BashOperator
-
-COMMANDE_BACKEND = "cd /opt/backend && env -u VIRTUAL_ENV uv run --no-sync python -m"
-
-CSV_PATH = "/opt/data/raw/all_sites_combined.csv"
-METADATA_PATH = "/opt/data/raw/dataset_metadata.json"
-SOURCE_TIMEZONE = "UTC"
-BATCH_SIZE = 1000
-
-with DAG(
-    dag_id="historical_import",
-    description="Importe le dataset historique CSV/JSON dans dataset, site et reading.",
-    schedule=None,
-    start_date=datetime(2026, 1, 1),
-    catchup=False,
-    max_active_runs=1,
-    tags=["etl", "historical"],
-) as dag:
-    BashOperator(
-        task_id="import_historical",
-        bash_command=(
-            f"{COMMANDE_BACKEND} app.etl.historical_import "
-            f"--csv {CSV_PATH} "
-            f"--metadata {METADATA_PATH} "
-            "--source-timezone UTC "
-            "--batch-size 1000"
-        ),
-        retries=1,
-        retry_delay=timedelta(minutes=2),
-        execution_timeout=timedelta(minutes=30),
-    )
+"""DAG d'import du dataset historique EnerVision (issue #119).
+
+Orchestre le pipeline existant `app.etl.historical_import` sans dupliquer sa logique ETL.
+Le dataset historique sert à initialiser l'environnement : le DAG reste donc manuel.
+
+Le backend est exécuté dans l'environnement `/opt/backend` embarqué dans l'image Airflow,
+sur le même patron que le DAG `alertes` (ADR 0008).
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+
+from airflow.providers.standard.operators.bash import BashOperator
+from airflow.sdk import DAG
+
+COMMANDE_BACKEND = "cd /opt/backend && env -u VIRTUAL_ENV uv run --no-sync python -m"
+
+CSV_PATH = "/opt/data/raw/all_sites_combined.csv"
+METADATA_PATH = "/opt/data/raw/dataset_metadata.json"
+SOURCE_TIMEZONE = "UTC"
+BATCH_SIZE = 1000
+
+with DAG(
+    dag_id="historical_import",
+    description="Importe le dataset historique CSV/JSON dans dataset, site et reading.",
+    schedule=None,
+    start_date=datetime(2026, 1, 1),
+    catchup=False,
+    max_active_runs=1,
+    tags=["etl", "historical"],
+) as dag:
+    BashOperator(
+        task_id="import_historical",
+        bash_command=(
+            f"{COMMANDE_BACKEND} app.etl.historical_import "
+            f"--csv {CSV_PATH} "
+            f"--metadata {METADATA_PATH} "
+            "--source-timezone UTC "
+            "--batch-size 1000"
+        ),
+        retries=1,
+        retry_delay=timedelta(minutes=2),
+        execution_timeout=timedelta(minutes=30),
+    )
diff --git a/etl/airflow/tests/test_dags.py b/etl/airflow/tests/test_dags.py
index 4d2c027..60978b5 100644
--- a/etl/airflow/tests/test_dags.py
+++ b/etl/airflow/tests/test_dags.py
@@ -49,7 +49,7 @@ def test_alertes_runs_after_the_hourly_scoring(dagbag: DagBag) -> None:
 
 
 def test_historical_import_has_no_schedule(dagbag: DagBag) -> None:
-    assert dagbag.dags["historical_import"].timetable.summary == "None"
+    assert dagbag.dags["historical_import"].schedule is None
 
 
 def test_ml_train_task_calls_the_training_module(dagbag: DagBag) -> None: