Merge pull request #1080 from MTES-MCT/staging

MEP v9

Author: alexisig

.env.example

@@ -52,3 +52,5 @@ DEBUG_TOOLBAR=1
 
 GOOGLE_ADWORDS_ACTIVATE=1
 DB_LOGGING_LEVEL=WARNING
+
+TWO_FACTOR_ENABLED=1

.github/workflows/deploy_airflow.yml

@@ -9,11 +9,19 @@ jobs:
     name: Deploy
     runs-on: ubuntu-latest
     steps:
-      - name: executing remote ssh git pull
+      - name: Git pull
         uses: appleboy/ssh-action@master
         with:
           host: ${{ secrets.AIRFLOW_SSH_HOST }}
           username: ${{ secrets.AIRFLOW_SSH_USER }}
           key: ${{ secrets.AIRFLOW_SSH_KEY }}
           port: ${{ secrets.AIRFLOW_SSH_PORT }}
           script: cd ~/sparte && git pull
+      - name: Generate DBT docs
+        uses: appleboy/ssh-action@master
+        with:
+            host: ${{ secrets.AIRFLOW_SSH_HOST }}
+            username: ${{ secrets.AIRFLOW_SSH_USER }}
+            key: ${{ secrets.AIRFLOW_SSH_KEY }}
+            port: ${{ secrets.AIRFLOW_SSH_PORT }}
+            script: cd ~/sparte/airflow/include/sql/sparte && dbt docs generate --static

.vscode/settings.json

@@ -34,5 +34,6 @@
     "[typescriptreact]": {
         "editor.defaultFormatter": "esbenp.prettier-vscode"
     },
+    "dbt.perspectiveTheme": "Vintage",
 }
 

Pipfile

@@ -71,3 +71,6 @@ setuptools = "*"
 py7zr = "*"
 dependency-injector = "*"
 django-webpack-loader = "*"
+django-two-factor-auth = "*"
+qrcode = "*"
+phonenumbers = "*"

Pipfile.lock

@@ -5,6 +5,7 @@
 from contextlib import contextmanager
 
 import pytest
+
 from airflow.hooks.base import BaseHook
 from airflow.models import Connection, DagBag, Variable
 from airflow.utils.db import initdb

airflow/.astro/test_dag_integrity_default.py

@@ -6,6 +6,5 @@ __pycache__/
 astro
 .venv
 airflow-webserver.pid
-webserver_config.py
 airflow.cfg
 airflow.db

airflow/.gitignore

@@ -0,0 +1 @@
+from .data_gouv import *  # noqa: F403, F401

airflow/dags/__init__.py

@@ -0,0 +1,65 @@
+from datetime import datetime, timedelta
+from os import getenv
+
+from include.container import Container
+
+from airflow import DAG
+from airflow.operators.bash import BashOperator
+from airflow.operators.python import PythonOperator
+
+default_args = {
+    "owner": "Alexis Athlani",
+    "depends_on_past": False,
+    "email_on_failure": True,
+    "email_on_retry": False,
+    "retries": 1,
+    "retry_delay": timedelta(minutes=5),
+}
+
+DB_HOST = getenv("DBT_DB_HOST")
+DB_PORT = getenv("DBT_DB_PORT")
+DB_NAME = getenv("DBT_DB_NAME")
+DB_USER = getenv("DBT_DB_USER")
+DB_PASSWORD = getenv("DBT_DB_PASSWORD")
+
+
+def upload_to_s3(local_file: str, s3_file: str) -> str:
+    bucket_name = "airflow-staging"
+    path_on_bucket = f"{bucket_name}/{s3_file}"
+
+    s3 = Container().s3()
+    s3.put(local_file, path_on_bucket)
+
+    print(f"Upload réussi : {path_on_bucket}")
+    return path_on_bucket
+
+
+with DAG(
+    "backup_dbt_staging_db",
+    default_args=default_args,
+    description="Sauvegarde hebdomadaire de la base staging DBT vers S3",
+    schedule_interval="0 1 * * 0",  # Tous les dimanches à 1h du matin
+    start_date=datetime(2024, 1, 1),
+    catchup=False,
+) as dag:
+    backup_date = "{{ ds_nodash }}"
+    backup_filename = f"{DB_NAME}_backup_{backup_date}.dump"
+    backup_path = f"/tmp/{backup_filename}"
+
+    create_backup = BashOperator(
+        task_id="create_backup",
+        bash_command=f'PGPASSWORD="{DB_PASSWORD}" pg_dump -Fc -O -h {DB_HOST} -p {DB_PORT} -U {DB_USER} -d {DB_NAME} > {backup_path}',  # noqa E501
+    )
+
+    upload_backup = PythonOperator(
+        task_id="upload_to_s3",
+        python_callable=upload_to_s3,
+        op_kwargs={"local_file": backup_path, "s3_file": f"backup/{DB_NAME}/{backup_filename}"},
+    )
+
+    cleanup = BashOperator(
+        task_id="cleanup",
+        bash_command=f"rm {backup_path}",
+    )
+
+    create_backup >> upload_backup >> cleanup

airflow/dags/backup_dbt_db.py

@@ -1,16 +1,23 @@
+import json
+
 import pendulum
+from include.domain.container import Container
+from include.utils import multiline_string_to_single_line
+
 from airflow.decorators import dag, task
 from airflow.exceptions import AirflowSkipException
 from airflow.models.param import Param
-from include.domain.container import Container
+
+with open("include/domain/data/ocsge/sources.json", "r") as f:
+    sources = json.load(f)
 
 
-def get_geojson_filename(year: int, departement: str) -> str:
-    return f"occupation_du_sol_{year}_{departement}.geojson"
+def get_geojson_filename(index: int, departement: str) -> str:
+    return f"occupation_du_sol_{index}_{departement}.geojson"
 
 
-def get_pmtiles_filename(year: int, departement: str) -> str:
-    return f"occupation_du_sol_{year}_{departement}.pmtiles"
+def get_pmtiles_filename(index: int, departement: str) -> str:
+    return f"occupation_du_sol_{index}_{departement}.pmtiles"
 
 
 @dag(
@@ -23,8 +30,8 @@ def get_pmtiles_filename(year: int, departement: str) -> str:
     tags=["OCS GE"],
     max_active_tasks=10,
     params={
-        "year": Param(2018, type="integer"),
-        "departement": Param("75", type="string"),
+        "index": Param(1, type="integer", enum=[1, 2]),
+        "departement": Param("75", type="string", enum=list(sources.keys())),
         "refresh_existing": Param(False, type="boolean"),
     },
 )
@@ -36,35 +43,55 @@ def create_ocsge_vector_tiles():
     def check_if_vector_tiles_not_exist(params: dict):
         if params.get("refresh_existing"):
             return
-        year = params.get("year")
+        index = params.get("index")
         departement = params.get("departement")
-        filename = get_pmtiles_filename(year, departement)
+        filename = get_pmtiles_filename(index, departement)
         exists = Container().s3().exists(f"{bucket_name}/{vector_tiles_dir}/{filename}")
         if exists:
             raise AirflowSkipException("Vector tiles already exist")
 
     @task.python(trigger_rule="none_skipped")
     def postgis_to_geojson(params: dict):
-        year = params.get("year")
+        index = params.get("index")
         departement = params.get("departement")
-        filename = get_geojson_filename(year, departement)
+        filename = get_geojson_filename(index, departement)
+
+        sql = f"""
+            SELECT
+                id,
+                code_cs,
+                code_us,
+                departement,
+                year,
+                index,
+                is_impermeable,
+                is_artificial,
+                critere_seuil,
+                st_transform(geom, 4326) as geom,
+                surface
+            FROM
+                public_ocsge.occupation_du_sol_with_artif
+            WHERE
+                index = {index} and
+                departement = '{departement}'
+        """
 
         return (
             Container()
             .sql_to_geojsonseq_on_s3_handler()
             .export_sql_result_to_geojsonseq_on_s3(
-                sql=f"SELECT * FROM public_ocsge.occupation_du_sol WHERE year = {year} and departement = '{departement}'",  # noqa: E501
+                sql=multiline_string_to_single_line(sql),  # noqa: E501
                 s3_key=f"{vector_tiles_dir}/{filename}",
                 s3_bucket=bucket_name,
             )
         )
 
     @task.bash(skip_on_exit_code=110, trigger_rule="none_skipped")
     def geojson_to_pmtiles(params: dict):
-        year = params.get("year")
+        index = params.get("index")
         departement = params.get("departement")
-        geojson_filename = get_geojson_filename(year, departement)
-        pmtiles_filename = get_pmtiles_filename(year, departement)
+        geojson_filename = get_geojson_filename(index, departement)
+        pmtiles_filename = get_pmtiles_filename(index, departement)
         local_input = f"/tmp/{geojson_filename}"
         local_output = f"/tmp/{pmtiles_filename}"
         Container().s3().get_file(f"{bucket_name}/{vector_tiles_dir}/{geojson_filename}", local_input)
@@ -87,25 +114,25 @@ def geojson_to_pmtiles(params: dict):
 
     @task.python(trigger_rule="none_skipped")
     def upload(params: dict):
-        year = params.get("year")
+        index = params.get("index")
         departement = params.get("departement")
-        pmtiles_filename = get_pmtiles_filename(year, departement)
+        pmtiles_filename = get_pmtiles_filename(index, departement)
         local_path = f"/tmp/{pmtiles_filename}"
         path_on_s3 = f"{bucket_name}/{vector_tiles_dir}/{pmtiles_filename}"
         Container().s3().put(local_path, path_on_s3)
 
     @task.bash(trigger_rule="none_skipped")
     def delete_geojson_file(params: dict):
-        year = params.get("year")
+        index = params.get("index")
         departement = params.get("departement")
-        geojson_filename = get_geojson_filename(year, departement)
+        geojson_filename = get_geojson_filename(index, departement)
         return f"rm /tmp/{geojson_filename}"
 
     @task.bash(trigger_rule="none_skipped")
     def delete_pmtiles_file(params: dict):
-        year = params.get("year")
+        index = params.get("index")
         departement = params.get("departement")
-        pmtiles_filename = get_pmtiles_filename(year, departement)
+        pmtiles_filename = get_pmtiles_filename(index, departement)
         return f"rm /tmp/{pmtiles_filename}"
 
     (