Ecrire toutes les tables DBT dans le schema warehouse

Author: kolok

.secrets.baseline

@@ -142,7 +142,7 @@
         "filename": "core/settings.py",
         "hashed_secret": "1ee34e26aeaf89c64ecc2c85efe6a961b75a50e9",
         "is_verified": false,
-        "line_number": 275
+        "line_number": 276
       }
     ],
     "dbt/package-lock.yml": [
@@ -191,5 +191,5 @@
       }
     ]
   },
-  "generated_at": "2025-04-24T11:22:55Z"
+  "generated_at": "2025-05-07T11:42:45Z"
 }

Makefile

@@ -146,7 +146,11 @@ dump-production:
 .PHONY: load-production-dump
 load-production-dump:
 	@DUMP_FILE=$$(find tmpbackup -type f -name "*.pgsql" -print -quit); \
-	pg_restore -d "$(DB_URL)" --clean --no-acl --no-owner --no-privileges "$$DUMP_FILE" || true
+	psql -d "$(DB_URL)" -c "CREATE EXTENSION IF NOT EXISTS postgis;" && \
+	psql -d "$(DB_URL)" -c 'CREATE EXTENSION IF NOT EXISTS "unaccent";' && \
+	psql -d "$(DB_URL)" -c 'CREATE EXTENSION IF NOT EXISTS "pg_trgm";' && \
+	psql -d "$(DB_URL)" -c 'CREATE EXTENSION IF NOT EXISTS "uuid-ossp";' && \
+	pg_restore -d "$(DB_URL)" --schema=public --clean --no-acl --no-owner --no-privileges "$$DUMP_FILE" || true
 	rm -rf tmpbackup
 
 .PHONY: db-restore

core/models/tools.py

@@ -5,7 +5,7 @@
 logger = logging.getLogger(__name__)
 
 
-def compare_model_vs_table(cls, table_name: str) -> bool:
+def compare_model_vs_table(cls, db_schema: str, table_name: str) -> bool:
 
     def are_fields_matched(db_type, model_type):
         logger.info(f"Comparing {db_type} with {model_type}")
@@ -26,8 +26,10 @@ def are_fields_matched(db_type, model_type):
     with connection.cursor() as cursor:
 
         table_info = connection.introspection.get_table_list(cursor)
+        for table in table_info:
+            logger.info(f"Table: {table}")
         if not any(table.name == table_name for table in table_info):
-            logger.error(f"La table {table_name} n'existe pas")
+            logger.error(f"La table {db_schema}.{table_name} n'existe pas")
             return False
 
         table_description = connection.introspection.get_table_description(
@@ -128,6 +130,8 @@ def are_fields_matched(db_type, model_type):
             through_model = field.remote_field.through
             through_table = through_model._meta.db_table
             through_table = through_table.replace(cls_table, table_name)
-            all_fields_match &= compare_model_vs_table(through_model, through_table)
+            all_fields_match &= compare_model_vs_table(
+                through_model, db_schema, through_table
+            )
 
         return all_fields_match

core/settings.py

@@ -266,6 +266,7 @@ def context_processors():
 )
 default_settings = dj_database_url.parse(DATABASE_URL)
 default_settings["ENGINE"] = "django.contrib.gis.db.backends.postgis"
+default_settings["OPTIONS"] = {"options": "-c search_path=public,warehouse"}
 
 # The readonly access is configured with fake access when DB_READONLY env
 # variable is not set.

dags/.env.template

@@ -59,4 +59,4 @@ POSTGRES_PORT=5432
 POSTGRES_USER=qfdmo
 POSTGRES_PASSWORD=qfdmo
 POSTGRES_DB=qfdmo
-POSTGRES_SCHEMA=public
+POSTGRES_SCHEMA=warehouse

dags/acteurs/dags/compute_acteur.py

@@ -123,11 +123,15 @@
     )
 
     check_model_table_displayedacteur_task = check_model_table_consistency_task(
-        dag, "DisplayedActeur", "exposure_carte_acteur"
+        dag, "qfdmo", "DisplayedActeur", "warehouse", "exposure_carte_acteur"
     )
     check_model_table_displayedpropositionservice_task = (
         check_model_table_consistency_task(
-            dag, "DisplayedPropositionService", "exposure_carte_propositionservice"
+            dag,
+            "qfdmo",
+            "DisplayedPropositionService",
+            "warehouse",
+            "exposure_carte_propositionservice",
         )
     )
     # The publication isn't handled by DBT yet
@@ -136,13 +140,17 @@
     # )
 
     check_model_table_vueacteur_task = check_model_table_consistency_task(
-        dag, "VueActeur", "exposure_exhaustive_acteur"
+        dag, "qfdmo", "VueActeur", "warehouse", "exposure_exhaustive_acteur"
     )
     check_model_table_vuepropositionservice_task = check_model_table_consistency_task(
-        dag, "VuePropositionService", "exposure_exhaustive_propositionservice"
+        dag,
+        "qfdmo",
+        "VuePropositionService",
+        "warehouse",
+        "exposure_exhaustive_propositionservice",
     )
     replace_vueacteur_table_task = replace_acteur_table_task(
-        dag, "qfdmo_vue", "exposure_exhaustive_"
+        dag, "public", "qfdmo_vue", "warehouse", "exposure_exhaustive_"
     )
 
     # Définir la séquence principale

dags/acteurs/tasks/airflow_logic/check_model_table_consistency_task.py

@@ -8,6 +8,7 @@
 from airflow import DAG
 from airflow.exceptions import AirflowFailException
 from airflow.operators.python import PythonOperator
+
 from utils import logging_utils as log
 
 logger = logging.getLogger(__name__)
@@ -30,30 +31,41 @@ def task_info_get(model_name, table_name):
 
 
 def check_model_table_consistency_wrapper(
-    ti, params, *, model_name: str, table_name: str
+    ti, params, *, django_app: str, model_name: str, db_schema: str, table_name: str
 ) -> None:
     # model_name = "DisplayedActeur"
     # table_name = "exposure_carte_acteur"
     logger.info(task_info_get(model_name, table_name))
 
     log.preview("Modèle Django", model_name)
     log.preview("Table", table_name)
-    if not check_model_table_consistency(model_name=model_name, table_name=table_name):
+    if not check_model_table_consistency(
+        django_app=django_app,
+        model_name=model_name,
+        db_schema=db_schema,
+        table_name=table_name,
+    ):
         raise AirflowFailException(
             f"le modèle {model_name} ne correspond pas à la table {table_name}"
         )
 
 
 def check_model_table_consistency_task(
-    dag: DAG, model_name: str, table_name: str
+    dag: DAG,
+    django_app: str,
+    model_name: str,
+    db_schema: str,
+    table_name: str,
 ) -> PythonOperator:
     task_name = f"check_{model_name}_vs_{table_name}_consistency"
     return PythonOperator(
         task_id=task_name,
         python_callable=check_model_table_consistency_wrapper,
         dag=dag,
         op_kwargs={
+            "django_app": django_app,
             "model_name": model_name,
+            "db_schema": db_schema,
             "table_name": table_name,
         },
     )

dags/acteurs/tasks/airflow_logic/replace_acteur_table_task.py

@@ -5,6 +5,7 @@
 from acteurs.tasks.business_logic.replace_acteur_table import replace_acteur_table
 from airflow import DAG
 from airflow.operators.python import PythonOperator
+
 from utils import logging_utils as log
 
 logger = logging.getLogger(__name__)
@@ -30,25 +31,42 @@ def task_info_get(prefix_django: str, prefix_dbt: str):
 
 
 def replace_acteur_table_wrapper(
-    ti, params, *, prefix_django: str, prefix_dbt: str
+    ti,
+    params,
+    *,
+    schema_django: str,
+    prefix_django: str,
+    schema_dbt: str,
+    prefix_dbt: str,
 ) -> None:
     logger.info(task_info_get(prefix_django, prefix_dbt))
 
     log.preview("Préfixe des tables du modèle Django", prefix_django)
     log.preview("Préfixe des tables calculées par DBT", prefix_dbt)
-    replace_acteur_table(prefix_django=prefix_django, prefix_dbt=prefix_dbt)
+    replace_acteur_table(
+        schema_django=schema_django,
+        prefix_django=prefix_django,
+        schema_dbt=schema_dbt,
+        prefix_dbt=prefix_dbt,
+    )
 
 
 def replace_acteur_table_task(
-    dag: DAG, prefix_django: str, prefix_dbt: str
+    dag: DAG,
+    schema_django: str,
+    prefix_django: str,
+    schema_dbt: str,
+    prefix_dbt: str,
 ) -> PythonOperator:
     task_name = f"replace_{prefix_django}_by_{prefix_dbt}_table"
     return PythonOperator(
         task_id=task_name,
         python_callable=replace_acteur_table_wrapper,
         dag=dag,
         op_kwargs={
+            "schema_django": schema_django,
             "prefix_django": prefix_django,
+            "schema_dbt": schema_dbt,
             "prefix_dbt": prefix_dbt,
         },
     )

dags/acteurs/tasks/business_logic/check_model_table_consistency.py

@@ -3,11 +3,17 @@
 django_setup_full()
 
 
-def check_model_table_consistency(*, model_name: str, table_name: str) -> bool:
+def check_model_table_consistency(
+    *,
+    django_app: str,
+    model_name: str,
+    db_schema: str,
+    table_name: str,
+) -> bool:
 
     from django.apps import apps
 
     from core.models.tools import compare_model_vs_table
 
-    model_class = apps.get_model("qfdmo", model_name)
-    return compare_model_vs_table(model_class, table_name)
+    model_class = apps.get_model(django_app, model_name)
+    return compare_model_vs_table(model_class, db_schema, table_name)

dags/acteurs/tasks/business_logic/replace_acteur_table.py

@@ -8,8 +8,10 @@
 
 
 def replace_acteur_table(
-    prefix_django="qfdmo_displayed",
-    prefix_dbt="exposure_carte_",
+    schema_django: str,
+    prefix_django: str,
+    schema_dbt: str,
+    prefix_dbt: str,
     tables=[
         "acteur",
         "acteur_acteur_services",
@@ -26,6 +28,15 @@ def replace_acteur_table(
         cursor.execute("BEGIN")
         try:
             for table in tables:
+                logger.warning(
+                    f"Update schema of {prefix_dbt}{table} from {schema_dbt}"
+                    f" to {schema_django}"
+                )
+                cursor.execute(
+                    f"ALTER TABLE {schema_dbt}.{prefix_dbt}{table}"
+                    f" SET SCHEMA {schema_django}"
+                )
+
                 logger.warning(
                     f"Renaming {prefix_django}{table}"
                     f" to {prefix_django}{table}_to_remove"