Ajout de tests de cohérences et de tags (#1611)

Author: kolok

dags/acteurs/dags/compute_acteur.py

@@ -10,6 +10,7 @@
 from airflow import DAG
 from airflow.operators.bash import BashOperator
 from shared.config.schedules import SCHEDULES
+from shared.config.tags import TAGS
 
 default_args = {
     "owner": "airflow",
@@ -31,6 +32,7 @@
         " (vue exhaustive des acteurs), par la carte (vue des acteurs affichés) et"
         " par l'export des acteurs en open-data."
     ),
+    tags=[TAGS.COMPUTE, TAGS.ACTEURS, TAGS.CARTE, TAGS.OPENDATA, TAGS.DBT],
     schedule=SCHEDULES.DAILY,
     max_active_runs=1,
 ) as dag:

dags/acteurs/dags/export_opendata.py

@@ -7,6 +7,7 @@
 from airflow import DAG
 from decouple import config
 from shared.config.schedules import SCHEDULES
+from shared.config.tags import TAGS
 
 ENVIRONMENT = config("ENVIRONMENT", default="development")
 
@@ -29,6 +30,7 @@
         "Ce DAG export les acteurs disponibles en opendata précédemment générés dans la"
         " table `exposure_opendata_acteur` de la base de données."
     ),
+    tags=[TAGS.COMPUTE, TAGS.EXPORT, TAGS.ACTEURS, TAGS.OPENDATA, TAGS.S3],
     params={
         "bucket_name": "lvao-opendata",
         "remote_dir": "acteurs" if ENVIRONMENT == "prod" else f"acteurs-{ENVIRONMENT}",

dags/clone/dags/clone_ae_etablissement.py

@@ -8,6 +8,7 @@
 from airflow.models.param import Param
 from clone.tasks.airflow_logic.chain_tasks import chain_tasks
 from shared.config import CATCHUPS, SCHEDULES, START_DATES
+from shared.config.tags import TAGS
 
 with DAG(
     dag_id="clone_ae_etablissement",
@@ -25,7 +26,13 @@
     description=(
         "Clone la table 'etablissement' de l'Annuaire Entreprises (AE) dans notre DB"
     ),
-    tags=["clone", "annuaire", "entreprise", "etablissement", "siret", "ae"],
+    tags=[
+        TAGS.ENRICH,
+        TAGS.CLONE,
+        TAGS.ANNAIRE_ENTREPRISE,
+        TAGS.ETABLISSEMENT,
+        TAGS.SIRET,
+    ],
     params={
         "dry_run": Param(
             False,

dags/clone/dags/clone_ae_unite_legale.py

@@ -8,6 +8,7 @@
 from airflow.models.param import Param
 from clone.tasks.airflow_logic.chain_tasks import chain_tasks
 from shared.config import CATCHUPS, SCHEDULES, START_DATES
+from shared.config.tags import TAGS
 
 with DAG(
     dag_id="clone_ae_unite_legale",
@@ -25,7 +26,13 @@
     description=(
         "Clone la table 'unite_legale' de l'Annuaire Entreprises (AE) dans notre DB"
     ),
-    tags=["clone", "annuaire", "entreprise", "unite_legale", "siren", "ae"],
+    tags=[
+        TAGS.ENRICH,
+        TAGS.CLONE,
+        TAGS.ANNAIRE_ENTREPRISE,
+        TAGS.UNITE_LEGALE,
+        TAGS.SIREN,
+    ],
     params={
         "dry_run": Param(
             False,

dags/clone/dags/clone_ban_adresses.py

@@ -8,6 +8,7 @@
 from airflow.models.param import Param
 from clone.tasks.airflow_logic.chain_tasks import chain_tasks
 from shared.config import CATCHUPS, SCHEDULES, START_DATES
+from shared.config.tags import TAGS
 
 with DAG(
     dag_id="clone_ban_adresses",
@@ -25,7 +26,7 @@
     description=(
         "Clone la table 'adresses' de la Base Adresse Nationale (BAN) dans notre DB"
     ),
-    tags=["clone", "ban", "adresses"],
+    tags=[TAGS.ENRICH, TAGS.CLONE, TAGS.BAN, TAGS.ADRESSES],
     params={
         "dry_run": Param(
             False,

dags/clone/dags/clone_ban_lieux_dits.py

@@ -8,6 +8,7 @@
 from airflow.models.param import Param
 from clone.tasks.airflow_logic.chain_tasks import chain_tasks
 from shared.config import CATCHUPS, SCHEDULES, START_DATES
+from shared.config.tags import TAGS
 
 with DAG(
     dag_id="clone_ban_lieux_dits",
@@ -25,7 +26,7 @@
     description=(
         "Clone la table 'lieux_dits' de la Base Adresse Nationale (BAN) dans notre DB"
     ),
-    tags=["clone", "ban", "adresses", "lieux_dits"],
+    tags=[TAGS.ENRICH, TAGS.CLONE, TAGS.BAN, TAGS.LIEUX_DITS],
     params={
         "dry_run": Param(
             False,

dags/cluster/dags/cluster_acteur_suggestions.py

@@ -7,6 +7,8 @@
 from cluster.tasks.airflow_logic.chain_tasks import chain_tasks
 from cluster.ui import params_separators as UI_PARAMS_SEPARATORS
 from shared.config import CATCHUPS, SCHEDULES, START_DATES
+from shared.config.tags import TAGS
+
 from utils.airflow_params import airflow_params_dropdown_from_mapping
 from utils.django import django_model_fields_get, django_setup_full
 
@@ -290,7 +292,7 @@
     catchup=CATCHUPS.AWLAYS_FALSE,
     schedule=SCHEDULES.NONE,
     description=("Un DAG pour générer des suggestions de clustering pour les acteurs"),
-    tags=["cluster", "acteurs", "suggestions"],
+    tags=[TAGS.CLUSTER, TAGS.ACTEURS, TAGS.SUGGESTIONS, TAGS.CLUSTERING],
     params=PARAMS,
 ) as dag:
     chain_tasks(dag)

dags/compute_acteurs/dags/create_final_actors.py

@@ -14,6 +14,7 @@
     deduplicate_labels_task,
     deduplicate_propositionservices_task,
 )
+from shared.config.tags import TAGS
 from shared.tasks.database_logic.db_tasks import read_data_from_postgres
 
 default_args = {
@@ -40,6 +41,7 @@
         " plusieurs sources en cumulant leur services, sources et propositions"
         " services."
     ),
+    tags=[TAGS.COMPUTE, TAGS.ACTEURS, TAGS.CARTE, TAGS.DEPRECATED],
     max_active_runs=1,
     schedule=None,
 )

dags/crawl/dags/crawl_urls.py

@@ -25,6 +25,7 @@
 from crawl.tasks.airflow_logic.crawl_urls_suggest_syntax_fail_task import (
     crawl_urls_suggest_syntax_fail_task,
 )
+from shared.config.tags import TAGS
 
 UI_PARAMS_SEPARATOR_SELECTION = r"""
 
@@ -50,7 +51,7 @@
     catchup=False,
     schedule_interval=None,
     description=("Un DAG pour parcourir des URLs et suggérer des corrections"),
-    tags=["crawl", "acteurs", "url", "suggestions"],
+    tags=[TAGS.ENRICH, TAGS.CRAWL, TAGS.ACTEURS, TAGS.URL, TAGS.SUGGESTIONS],
     params={
         "dry_run": Param(
             True,

dags/enrich/dags/enrich_acteurs_closed.py

@@ -18,6 +18,7 @@
     enrich_dbt_models_refresh_task,
 )
 from shared.config import CATCHUPS, SCHEDULES, START_DATES, config_to_airflow_params
+from shared.config.tags import TAGS
 
 with DAG(
     dag_id="enrich_acteurs_closed",
@@ -33,7 +34,14 @@
         "Un DAG pour détécter et remplacer les acteurs fermés"
         "dans l'Annuaire Entreprises (AE)"
     ),
-    tags=["annuaire", "entreprises", "ae", "siren", "siret", "acteurs", "fermés"],
+    tags=[
+        TAGS.ENRICH,
+        TAGS.ANNAIRE_ENTREPRISE,
+        TAGS.SIREN,
+        TAGS.SIRET,
+        TAGS.ACTEURS,
+        TAGS.CLOSED,
+    ],
     schedule=SCHEDULES.NONE,
     catchup=CATCHUPS.AWLAYS_FALSE,
     start_date=START_DATES.YESTERDAY,