Merge branch 'master' into alice

Author: aupaysdesdata

.github/workflows/ci.yml

@@ -0,0 +1,39 @@
+name: CI
+
+on:
+  push:
+    branches: ["**"]
+  pull_request:
+    branches: [master]
+
+jobs:
+  compile-check:
+    name: Vérification compilation / syntaxe
+    if: github.ref != 'refs/heads/master'
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout du code
+        uses: actions/checkout@v5
+
+      - name: Setup Python
+        uses: actions/setup-python@v6
+        with:
+          python-version: "3.12"
+
+      - name: Vérifier que tous les fichiers .py du src compilent
+        run: python -m py_compile $(find src -name "*.py")
+
+  docker-build:
+    name: Test de build du container
+    if: github.event_name == 'pull_request'
+    runs-on: ubuntu-latest
+    needs: compile-check
+    steps:
+      - name: Checkout du code
+        uses: actions/checkout@v5
+
+      - name: Setup Docker Buildx
+        uses: docker/setup-buildx-action@v3
+
+      - name: Build de toutes les images via docker-compose
+        run: docker compose build

dags/pipeline_accidents.py

@@ -3,25 +3,32 @@
 from airflow.providers.docker.operators.docker import DockerOperator
 from airflow.operators.python import PythonOperator
 from airflow.exceptions import AirflowFailException
+from docker.types import Mount
 import mlflow
+from airflow.utils.trigger_rule import TriggerRule
 from mlflow.tracking import MlflowClient
 import requests
 import os
+import re
+from pathlib import Path
 
 
+DATA_VOLUME_NAME = os.getenv("DATA_VOLUME_NAME", "mlops_accidents_accidents-data")
 MLFLOW_TRACKING_URI = "http://mlflow:5000"
 EXPERIMENT_NAME = "Gravité_Accidents"
 MODEL_NAME = "Modèle_Gravité_Accidents"
 DOCKER_NETWORK = "mlops_accidents_default"
 
 MAX_DROP_ALLOWED = 0.05
 
+
 default_args = {
-    'owner': 'mlops_team',
-    'start_date': datetime(2026, 1, 1),
-    'retries': 1,
+    "owner": "mlops_team",
+    "start_date": datetime(2026, 1, 1),
+    "retries": 1,
 }
 
+
 def check_metrics_and_alert(**context):
     """
     Récupère le f1_score du dernier run de l'expérience 'Gravité_Accidents'
@@ -32,26 +39,34 @@ def check_metrics_and_alert(**context):
 
     experiment = client.get_experiment_by_name(EXPERIMENT_NAME)
     if not experiment:
-        raise AirflowFailException(f"L'expérience '{EXPERIMENT_NAME}' n'a pas été trouvée dans MLflow.")
+        raise AirflowFailException(
+            f"L'expérience '{EXPERIMENT_NAME}' n'a pas été trouvée dans MLflow."
+        )
 
     runs = client.search_runs(
         experiment_ids=[experiment.experiment_id],
         order_by=["attributes.start_time DESC"],
-        max_results=1
+        max_results=1,
     )
 
     if not runs:
-        raise AirflowFailException(f"Aucun run trouvé pour l'expérience {EXPERIMENT_NAME}.")
+        raise AirflowFailException(
+            f"Aucun run trouvé pour l'expérience {EXPERIMENT_NAME}."
+        )
 
     current_run = runs[0]
     current_f1 = current_run.data.metrics.get("f1_score")
     current_run_id = current_run.info.run_id
 
-    context['ti'].xcom_push(key='current_run_id', value=current_run_id)
-    print(f"Nouveau modèle entraîné détecté - Run ID: {current_run_id} | F1-Score: {current_f1}")
+    context["ti"].xcom_push(key="current_run_id", value=current_run_id)
+    print(
+        f"Nouveau modèle entraîné détecté - Run ID: {current_run_id} | F1-Score: {current_f1}"
+    )
 
     if current_f1 is None:
-        raise AirflowFailException("Le dernier run MLflow n'a pas enregistré de métrique 'f1_score'.")
+        raise AirflowFailException(
+            "Le dernier run MLflow n'a pas enregistré de métrique 'f1_score'."
+        )
 
     try:
         prod_model_version = client.get_model_version_by_alias(MODEL_NAME, "champion")
@@ -67,6 +82,7 @@ def check_metrics_and_alert(**context):
     except mlflow.exceptions.MlflowException:
         print("Aucun modèle marqué '@champion' trouvé. Première promotion du projet.")
 
+
 def promote_model_to_champion(**context):
     """
     Associe l'alias 'champion' à la dernière version du modèle validé.
@@ -75,18 +91,22 @@ def promote_model_to_champion(**context):
     mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)
     client = MlflowClient()
 
-    run_id = context['ti'].xcom_pull(key='current_run_id', task_ids='evaluate_metrics')
+    run_id = context["ti"].xcom_pull(key="current_run_id", task_ids="evaluate_metrics")
 
     filter_string = f"run_id='{run_id}'"
     versions = client.search_model_versions(filter_string)
 
     if not versions:
-        raise AirflowFailException(f"Aucune version de modèle trouvée dans le Registry pour le run {run_id}.")
+        raise AirflowFailException(
+            f"Aucune version de modèle trouvée dans le Registry pour le run {run_id}."
+        )
 
     latest_version = versions[0].version
 
     client.set_registered_model_alias(MODEL_NAME, "champion", latest_version)
-    print(f"Succès : Le modèle '{MODEL_NAME}' version {latest_version} est maintenant désigné comme '@champion'.")
+    print(
+        f"Succès : Le modèle '{MODEL_NAME}' version {latest_version} est maintenant désigné comme '@champion'."
+    )
 
 
 def reload_predict_service():
@@ -97,23 +117,20 @@ def reload_predict_service():
     bento_url = "http://ml-api:3000/reload_model"
     try:
         response = requests.post(bento_url, timeout=15)
-        if response.status_code == 200:
-            print("Le conteneur 'ml-api' a mis à jour son modèle avec succès.")
-        else:
-            print(f"Le service ml-api a répondu avec un code erreur : {response.status_code}")
+        response.raise_for_status()
+        print("Le conteneur 'ml-api' a mis à jour son modèle avec succès.")
     except Exception as e:
-        print(f"Notification non envoyée à ml-api (Vérifie si l'API expose ce endpoint) : {e}")
+        raise AirflowFailException(f"Échec du rechargement de ml-api : {e}")
 
 
 with DAG(
-    'mlops_accident_gravity_pipeline',
+    "mlops_accident_gravity_pipeline",
     default_args=default_args,
-    description='Pipeline d\'entraînement pour la gravité des accidents',
-    schedule='@monthly',
+    description="Pipeline d'entraînement pour la gravité des accidents",
+    schedule="@monthly",
     catchup=False,
     tags=["accidents"],
 ) as dag:
-
     # task_make_dataset = DockerOperator(
     #     task_id='docker_make_dataset',
     #     image='make_dataset:latest',
@@ -123,28 +140,43 @@ def reload_predict_service():
     #     mounts=[Mount(source=f"{BASE_DIR}/mlruns", target="/app/mlruns", type="bind")]
     # )
 
+    task_preprocess = DockerOperator(
+        task_id="preprocess",
+        image="mlops_accidents-preprocess:latest",
+        api_version="auto",
+        auto_remove=True,
+        mount_tmp_dir=False,
+        network_mode=DOCKER_NETWORK,
+        mounts=[Mount(source=f"{DATA_VOLUME_NAME}", target="/app/data", type="volume")],
+    )
+
     task_train = DockerOperator(
-        task_id='docker_train',
-        image='mlops_accidents-train:latest',
-        api_version='auto',
+        task_id="train",
+        image="mlops_accidents-train:latest",
+        api_version="auto",
         auto_remove=True,
+        mount_tmp_dir=False,
         network_mode=DOCKER_NETWORK,
-        environment={'MLFLOW_TRACKING_URI': MLFLOW_TRACKING_URI},
+        environment={"MLFLOW_TRACKING_URI": MLFLOW_TRACKING_URI},
+        mounts=[Mount(source=f"{DATA_VOLUME_NAME}", target="/app/data", type="volume")],
     )
 
     task_evaluate = PythonOperator(
-        task_id='evaluate_metrics',
+        task_id="evaluate_metrics",
         python_callable=check_metrics_and_alert,
     )
 
     task_promote = PythonOperator(
-        task_id='promote_model',
+        task_id="promote_model",
         python_callable=promote_model_to_champion,
     )
 
     task_reload = PythonOperator(
-        task_id='reload_predict_service',
-        python_callable=reload_predict_service
+        task_id="reload_predict_service",
+        python_callable=reload_predict_service,
+        trigger_rule=TriggerRule.ALL_DONE,
     )
 
-    task_train >> task_evaluate >> task_promote >> task_reload
+    task_preprocess >> task_train >> task_evaluate
+    task_evaluate >> task_promote
+    task_evaluate >> task_reload

docker-compose.yml

@@ -1,4 +1,3 @@
-# version: '3.8'
 services:
   mlflow:
     image: ghcr.io/mlflow/mlflow:v2.11.3
@@ -7,6 +6,14 @@ services:
     command: mlflow server --backend-store-uri sqlite:///mlflow.db --artifacts-destination /app/mlruns --serve-artifacts --host 0.0.0.0 --port 5000
     volumes:
       - ./mlruns:/app/mlruns
+  
+  preprocess:
+    build:
+      context: .
+      dockerfile: src/preprocess/Dockerfile
+    volumes:
+      - accidents-data:/app/data
+    restart: "no"
 
   train:
     build:
@@ -16,6 +23,12 @@ services:
       - MLFLOW_TRACKING_URI=http://mlflow:5000
     volumes:
       - ./mlruns:/app/mlruns
+    depends_on:
+      preprocess:
+        condition: service_completed_successfully
+      mlflow:
+        condition: service_started
+    restart: "no"
 
   ml-api:
     build:
@@ -39,10 +52,11 @@ services:
       dockerfile: src/streamlit/Dockerfile
     environment:
       - MODEL_API_URL=http://ml-api:3000
-    expose:
+    ports:
       - "8501"
     depends_on:
-      - ml-api
+      ml-api:
+        condition: service_started
 
   nginx:
     build:
@@ -52,8 +66,10 @@ services:
       - "443:443"
       - "80:80"
     depends_on:
-      - streamlit
-      - ml-api
+      streamlit:
+        condition: service_started
+      ml-api:
+        condition: service_started
     restart: unless-stopped
 
   prometheus:
@@ -67,7 +83,8 @@ services:
       - '--config.file=/etc/prometheus/prometheus.yml'
       - '--storage.tsdb.path=/prometheus'
     depends_on:
-      - ml-api
+      ml-api:
+        condition: service_started
 
   grafana:
     image: grafana/grafana:latest
@@ -81,7 +98,8 @@ services:
     ports:
       - "3000:3000"
     depends_on:
-      - prometheus
+      prometheus:
+        condition: service_started
 
   postgres-airflow:
     image: postgres:15
@@ -93,10 +111,28 @@ services:
       - postgres-airflow-data:/var/lib/postgresql/data
     healthcheck:
       test: ["CMD", "pg_isready", "-U", "airflow"]
-      interval: 10s
+      interval: 5s
       timeout: 5s
       retries: 5
 
+  airflow-init:
+    build:
+      context: .
+      dockerfile: Dockerfile.airflow
+    environment:
+      AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres-airflow/airflow
+    user: root
+    command: >
+      bash -c "
+        su -s /bin/bash airflow -c '
+          airflow db migrate &&
+          airflow users create --username admin --password admin --firstname Admin --lastname Admin --role Admin --email admin@example.com || true
+        '
+      "
+    depends_on:
+      postgres-airflow:
+        condition: service_healthy
+
   airflow:
     build:
       context: .
@@ -115,36 +151,33 @@ services:
     volumes:
       - ./dags:/opt/airflow/dags
       - ./src:/opt/airflow/src
-      - ./data:/opt/airflow/data
+      - accidents-data:/opt/airflow/data
       - airflow-logs:/opt/airflow/logs
       - /var/run/docker.sock:/var/run/docker.sock
     ports:
       - "8080:8080"
-    
     user: root
-    # user: "0:0"
-    
     command: >
       bash -c "
         chmod 666 /var/run/docker.sock || true &&
         rm -f /opt/airflow/airflow-webserver.pid &&
         su -s /bin/bash airflow -c '
-          airflow db init &&
-          airflow users create --username admin --password admin --firstname Admin --lastname Admin --role Admin --email admin@example.com || true &&
           airflow scheduler &
-          airflow webserver
+          exec airflow webserver
         '
       "
-
     depends_on:
       postgres-airflow:
         condition: service_healthy
+      airflow-init:
+        condition: service_completed_successfully
       mlflow:
         condition: service_started
     restart: on-failure
 
 volumes:
   postgres-airflow-data:
   airflow-logs:
+  accidents-data:
   prometheus_data:
   grafana_data: