reorganisation par layer

dags/acteur_views/dags/build_vue.py

@@ -25,51 +25,68 @@
     schedule=None,
     max_active_runs=1,
 ) as dag:
+    """
+    # Exhaustive
+    dbt run --models base
+    dbt test --models base
+    dbt run --models intermediate
+    dbt test --models intermediate
+    dbt test --models intermediate
+    dbt test --models marts.exhaustive
+    dbt run --models marts.exhaustive
+    dbt test --models marts.exhaustive
+    dbt run --models exposure.exhaustive
+    dbt test --models exposure.exhaustive
 
-    run_dbt_exhaustive_acteurs_model = BashOperator(
-        task_id="build_exhaustive_acteurs",
-        bash_command=(
-            "cd /opt/airflow/dbt/ && dbt run --select qfdmo.exhaustive_acteurs"
-        ),
+    """
+    dbt_run_base = BashOperator(
+        task_id="run_base",
+        bash_command=("cd /opt/airflow/dbt/ && dbt run --models base"),
         dag=dag,
     )
-    test_dbt_exhaustive_acteurs_model = BashOperator(
-        task_id="test_exhaustive_acteurs",
-        bash_command=(
-            "cd /opt/airflow/dbt/ && dbt test --select qfdmo.exhaustive_acteurs"
-        ),
+    dbt_test_base = BashOperator(
+        task_id="test_base",
+        bash_command=("cd /opt/airflow/dbt/ && dbt test --models base"),
         dag=dag,
     )
-    run_dbt_carte_acteurs_model = BashOperator(
-        task_id="build_carte_acteurs",
-        bash_command=("cd /opt/airflow/dbt/ && dbt run --select qfdmo.carte_acteurs"),
+    dbt_run_intermediate = BashOperator(
+        task_id="run_intermediate",
+        bash_command=("cd /opt/airflow/dbt/ && dbt run --models intermediate"),
         dag=dag,
     )
-    test_dbt_carte_acteurs_model = BashOperator(
-        task_id="test_carte_acteurs",
-        bash_command=("cd /opt/airflow/dbt/ && dbt test --select qfdmo.carte_acteurs"),
+    dbt_test_intermediate = BashOperator(
+        task_id="test_intermediate",
+        bash_command=("cd /opt/airflow/dbt/ && dbt test --models intermediate"),
         dag=dag,
     )
-    run_dbt_opendata_acteurs_model = BashOperator(
-        task_id="build_opendata_acteurs",
-        bash_command=(
-            "cd /opt/airflow/dbt/ && dbt run --select qfdmo.opendata_acteurs"
-        ),
+    dbt_run_marts_exhaustive = BashOperator(
+        task_id="run_marts_exhaustive",
+        bash_command=("cd /opt/airflow/dbt/ && dbt run --models marts.exhaustive"),
         dag=dag,
     )
-    test_dbt_opendata_acteurs_model = BashOperator(
-        task_id="test_opendata_acteurs",
-        bash_command=(
-            "cd /opt/airflow/dbt/ && dbt test --select qfdmo.opendata_acteurs"
-        ),
+    dbt_test_marts_exhaustive = BashOperator(
+        task_id="test_marts_exhaustive",
+        bash_command=("cd /opt/airflow/dbt/ && dbt test --models marts.exhaustive"),
+        dag=dag,
+    )
+    dbt_run_exposure_exhaustive = BashOperator(
+        task_id="run_exposure_exhaustive",
+        bash_command=("cd /opt/airflow/dbt/ && dbt run --models exposure.exhaustive"),
+        dag=dag,
+    )
+    dbt_test_exposure_exhaustive = BashOperator(
+        task_id="test_exposure_exhaustive",
+        bash_command=("cd /opt/airflow/dbt/ && dbt test --models exposure.exhaustive"),
         dag=dag,
     )
 
     (
-        run_dbt_exhaustive_acteurs_model
-        >> test_dbt_exhaustive_acteurs_model
-        >> run_dbt_carte_acteurs_model
-        >> test_dbt_carte_acteurs_model
-        >> run_dbt_opendata_acteurs_model
-        >> test_dbt_opendata_acteurs_model
+        dbt_run_base
+        >> dbt_test_base
+        >> dbt_run_intermediate
+        >> dbt_test_intermediate
+        >> dbt_run_marts_exhaustive
+        >> dbt_test_marts_exhaustive
+        >> dbt_run_exposure_exhaustive
+        >> dbt_test_exposure_exhaustive
     )

dbt/dbt_project.yml

@@ -23,12 +23,29 @@ clean-targets:
   - "dbt_packages"
 
 models:
-  qfdmo:
-    exhaustive_acteurs:
-      materialized: table
-
-    opendata_acteurs:
-      materialized: table
-
-    carte_acteurs:
-      materialized: table
+  base:
+    schema: public
+    +materialized: view
+  intermediate:
+    schema: public
+    +materialized: view
+  marts:
+    exhaustive:
+      schema: public
+      +materialized: view
+    carte:
+      schema: public
+      +materialized: view
+    opendata:
+      schema: public
+      +materialized: view
+  exposure:
+    exhaustive:
+      schema: public
+      +materialized: view
+    carte:
+      schema: public
+      +materialized: view
+    opendata:
+      schema: public
+      +materialized: view

dbt/macros/macro_acteur.sql

@@ -0,0 +1,8 @@
+{%- macro acteur(ephemeral_filtered_acteur, propositionservice ) -%}
+
+SELECT DISTINCT va.*
+FROM {{ ref(ephemeral_filtered_acteur) }} AS va
+INNER JOIN {{ ref(propositionservice) }} AS cps
+    ON va.identifiant_unique = cps.acteur_id
+
+{%- endmacro -%}

dbt/models/opendata_acteurs/opendata_acteur_acteur_services.sql → dbt/macros/macro_acteur_acteur_services.sql

@@ -1,17 +1,19 @@
+{%- macro acteur_acteur_services(ephemeral_filtered_acteur, acteur ) -%}
+
 WITH nochild_acteur_acteur_services AS (
     SELECT
         aas.vueacteur_id AS acteur_id,
         aas.acteurservice_id AS acteurservice_id
     FROM qfdmo_vueacteur_acteur_services aas
-    INNER JOIN {{ ref('temp_opendata_filteredacteur') }} AS a ON aas.vueacteur_id = a.identifiant_unique AND a.parent_id is null
+    INNER JOIN {{ ref(ephemeral_filtered_acteur) }} AS a ON aas.vueacteur_id = a.identifiant_unique AND a.parent_id is null
     GROUP BY aas.vueacteur_id, aas.acteurservice_id
 ),
 parentacteur_acteur_services AS (
     SELECT
         a.parent_id AS acteur_id,
         aas.acteurservice_id AS acteurservice_id
     FROM qfdmo_vueacteur_acteur_services aas
-    INNER JOIN {{ ref('temp_opendata_filteredacteur') }} AS a ON aas.vueacteur_id = a.identifiant_unique AND a.parent_id is not null
+    INNER JOIN {{ ref(ephemeral_filtered_acteur) }} AS a ON aas.vueacteur_id = a.identifiant_unique AND a.parent_id is not null
     GROUP BY a.parent_id, aas.acteurservice_id
 ),
 acteur_acteur_services AS (
@@ -22,4 +24,6 @@ acteur_acteur_services AS (
 
 SELECT ROW_NUMBER() OVER (ORDER BY acteur_id, aas.acteurservice_id) AS id, aas.*
 FROM acteur_acteur_services AS aas
-INNER JOIN {{ ref('opendata_acteur') }} AS a ON a.identifiant_unique = acteur_id
+INNER JOIN {{ ref(acteur) }} AS a ON a.identifiant_unique = acteur_id
+
+{%- endmacro -%}

dbt/models/opendata_acteurs/opendata_acteur_labels.sql → dbt/macros/macro_acteur_labels.sql

@@ -1,17 +1,19 @@
+{%- macro acteur_labels(ephemeral_filtered_acteur, acteur ) -%}
+
 WITH nochild_acteur_labels AS (
     SELECT
         al.vueacteur_id AS acteur_id,
         al.labelqualite_id AS labelqualite_id
     FROM qfdmo_vueacteur_labels al
-    INNER JOIN {{ ref('temp_opendata_filteredacteur') }} AS a ON al.vueacteur_id = a.identifiant_unique AND a.parent_id is null
+    INNER JOIN {{ ref(ephemeral_filtered_acteur) }} AS a ON al.vueacteur_id = a.identifiant_unique AND a.parent_id is null
     GROUP BY al.vueacteur_id, al.labelqualite_id
 ),
 parentacteur_labels AS (
     SELECT
         a.parent_id AS acteur_id,
         al.labelqualite_id AS labelqualite_id
     FROM qfdmo_vueacteur_labels al
-    INNER JOIN {{ ref('temp_opendata_filteredacteur') }} AS a ON al.vueacteur_id = a.identifiant_unique AND a.parent_id is not null
+    INNER JOIN {{ ref(ephemeral_filtered_acteur) }} AS a ON al.vueacteur_id = a.identifiant_unique AND a.parent_id is not null
     GROUP BY a.parent_id, al.labelqualite_id
 ),
 acteur_labels AS (
@@ -22,4 +24,6 @@ acteur_labels AS (
 
 SELECT ROW_NUMBER() OVER (ORDER BY acteur_id, al.labelqualite_id) AS id, al.*
 FROM acteur_labels AS al
-INNER JOIN {{ ref('opendata_acteur') }} AS a ON a.identifiant_unique = acteur_id
+INNER JOIN {{ ref(acteur) }} AS a ON a.identifiant_unique = acteur_id
+
+{%- endmacro -%}

dbt/models/opendata_acteurs/opendata_acteur_sources.sql → dbt/macros/macro_acteur_sources.sql

@@ -1,16 +1,18 @@
+{%- macro acteur_sources(ephemeral_filtered_acteur, acteur ) -%}
+
 WITH nochild_acteur_labels AS (
     SELECT
         a.identifiant_unique AS acteur_id,
         a.source_id AS source_id
-    FROM {{ ref('temp_opendata_filteredacteur') }} AS a
+    FROM {{ ref(ephemeral_filtered_acteur) }} AS a
     WHERE a.parent_id is null AND a.source_id is not null
     GROUP BY a.identifiant_unique, a.source_id
 ),
 parentacteur_labels AS (
     SELECT
         a.parent_id AS acteur_id,
         a.source_id AS source_id
-    FROM {{ ref('temp_opendata_filteredacteur') }} AS a
+    FROM {{ ref(ephemeral_filtered_acteur) }} AS a
     WHERE a.parent_id is not null
     GROUP BY a.parent_id, a.source_id
 ),
@@ -22,4 +24,6 @@ acteur_sources AS (
 
 SELECT ROW_NUMBER() OVER (ORDER BY acteur_id, s.source_id) AS id, s.*
 FROM acteur_sources AS s
-INNER JOIN {{ ref('opendata_acteur') }} AS a ON a.identifiant_unique = acteur_id
+INNER JOIN {{ ref(acteur) }} AS a ON a.identifiant_unique = acteur_id
+
+{%- endmacro -%}

dbt/models/opendata_acteurs/temp_opendata_parentpropositionservice.sql → dbt/macros/macro_filtered_parentpropositionservice.sql

@@ -1,10 +1,13 @@
+{%- macro filtered_parentpropositionservice(ephemeral_filtered_acteur ) -%}
 
 SELECT
     tvps.id AS id,
     tcfa.parent_id AS parent_id,
     tvps.acteur_id AS acteur_id,
     tvps.action_id AS action_id
 FROM qfdmo_vuepropositionservice AS tvps
-    INNER JOIN {{ ref('temp_opendata_filteredacteur') }} AS tcfa
+    INNER JOIN {{ ref(ephemeral_filtered_acteur) }} AS tcfa
         ON tvps.acteur_id = tcfa.identifiant_unique
             AND tcfa.parent_id IS NOT NULL
+
+{%- endmacro -%}

dbt/models/carte_acteurs/temp_propositionservice.sql → dbt/macros/macro_filtered_propositionservice.sql

@@ -1,9 +1,11 @@
+{%- macro filtered_propositionservice(ephemeral_filtered_acteur, ephemeral_filtered_parentpropositionservice ) -%}
+
 with parent_propositionservice AS (
     SELECT
     concat(pps.parent_id::text, '_', pps.action_id::text) AS id,
     pps.parent_id AS acteur_id,
     pps.action_id AS action_id
-    FROM {{ ref('temp_parentpropositionservice') }} AS pps
+    FROM {{ ref(ephemeral_filtered_parentpropositionservice) }} AS pps
     group by 2,3
 ),
 nochild_propositionservice AS (
@@ -12,10 +14,12 @@ nochild_propositionservice AS (
     vps.acteur_id AS acteur_id,
     vps.action_id AS action_id
     FROM qfdmo_vuepropositionservice AS vps
-    INNER JOIN {{ ref('temp_filteredacteur') }} AS cfa
+    INNER JOIN {{ ref(ephemeral_filtered_acteur) }} AS cfa
         ON vps.acteur_id = cfa.identifiant_unique AND cfa.parent_id is null
 )
 
 SELECT * FROM parent_propositionservice
 union all
 SELECT * FROM nochild_propositionservice
+
+{%- endmacro -%}

dbt/macros/macro_propositionservice.sql

@@ -0,0 +1,12 @@
+{%- macro propositionservice(ephemeral_filtered_propositionservice, propositionservice_sous_categories ) -%}
+
+SELECT
+    MIN(ps.id) AS id,
+    ps.acteur_id,
+    ps.action_id
+FROM {{ ref(ephemeral_filtered_propositionservice) }} AS ps
+INNER JOIN {{ ref(propositionservice_sous_categories) }} AS pssscat
+   ON ps.id = pssscat.propositionservice_id
+GROUP BY acteur_id, action_id
+
+{%- endmacro -%}

dbt/macros/macro_propositionservice_sous_categories.sql

@@ -0,0 +1,37 @@
+{%- macro propositionservice_sous_categories(ephemeral_filtered_acteur, ephemeral_filtered_propositionservice, ephemeral_filtered_parentpropositionservice ) -%}
+
+with
+    parent_propositionservice_sous_categories
+    AS
+    (
+        SELECT
+            MIN(pssscat.id) AS id,
+            CONCAT(pps.parent_id::text, '_', pps.action_id::text) AS propositionservice_id,
+            pssscat.souscategorieobjet_id AS souscategorieobjet_id
+        FROM {{ ref("int_propositionservice_sous_categories") }} AS pssscat
+            INNER JOIN {{ ref(ephemeral_filtered_parentpropositionservice) }} AS pps
+                ON pps.id = pssscat.propositionservice_id
+        GROUP BY
+            pps.parent_id,
+            pps.action_id,
+            pssscat.souscategorieobjet_id
+    ),
+    nochild_propositionservice_sous_categories
+    AS
+    (
+        SELECT
+            pssscat.id AS id,
+            pssscat.propositionservice_id AS propositionservice_id,
+            pssscat.souscategorieobjet_id AS souscategorieobjet_id
+        FROM {{ ref("int_propositionservice_sous_categories") }} AS pssscat
+            INNER JOIN {{ ref(ephemeral_filtered_propositionservice) }} AS ps ON pssscat.propositionservice_id = ps.id
+            INNER JOIN {{ ref(ephemeral_filtered_acteur) }} AS cfa ON ps.acteur_id = cfa.identifiant_unique AND cfa.parent_id is null
+    )
+
+SELECT *
+    FROM parent_propositionservice_sous_categories
+UNION ALL
+SELECT *
+    FROM nochild_propositionservice_sous_categories
+
+{%- endmacro -%}

dbt/models/base/base_acteur.sql

@@ -0,0 +1 @@
+select * from {{ source('qfdmo', 'qfdmo_acteur') }}

dbt/models/base/base_acteur_acteur_services.sql

@@ -0,0 +1 @@
+select * from {{ source('qfdmo', 'qfdmo_acteur_acteur_services') }}

dbt/models/base/base_acteur_labels.sql

@@ -0,0 +1 @@
+select * from {{ source('qfdmo', 'qfdmo_acteur_labels') }}

dbt/models/base/base_propositionservice.sql

@@ -0,0 +1 @@
+select * from {{ source('qfdmo', 'qfdmo_propositionservice') }}

dbt/models/base/base_propositionservice_sous_categories.sql

@@ -0,0 +1 @@
+select * from {{ source('qfdmo', 'qfdmo_propositionservice_sous_categories') }}

dbt/models/base/base_revisionacteur.sql

@@ -0,0 +1 @@
+select * from {{ source('qfdmo', 'qfdmo_revisionacteur') }}

dbt/models/base/base_revisionacteur_acteur_services.sql

@@ -0,0 +1 @@
+select * from {{ source('qfdmo', 'qfdmo_revisionacteur_acteur_services') }}

dbt/models/base/base_revisionacteur_labels.sql

@@ -0,0 +1 @@
+select * from {{ source('qfdmo', 'qfdmo_revisionacteur_labels') }}

dbt/models/base/base_revisionpropositionservice.sql

@@ -0,0 +1 @@
+select * from {{ source('qfdmo', 'qfdmo_revisionpropositionservice') }}

dbt/models/base/base_revisionpropositionservice_sous_categories.sql

@@ -0,0 +1 @@
+select * from {{ source('qfdmo', 'qfdmo_revisionpropositionservice_sous_categories') }}

dbt/models/base/base_source.sql

@@ -0,0 +1 @@
+select * from {{ source('qfdmo', 'qfdmo_source') }}