chore(pipeline): add final dimension tables

The following changes are scoped to the final models:

* clean up unmaintained doc
* rename `marts/api` to `marts/inclusion`
* rename models to following naming convention
* force full-refresh for schema seeds
* add unique constraint to schema seeds to be able to set foreign keys
* unnest jsonb list columns to new tables, with contracts
* enforce contract for marts by default
* set indexes using dbt contracts
* set foreign keys using dbt contracts

pipeline/dags/dag_utils/marts.py

@@ -23,7 +23,7 @@ def _export_di_dataset_to_s3(logical_date, run_id):
         # TODO(vmttn): normalize table name after db split
         table_name = ressource.rstrip("s")
         key = f"{prefix}{ressource}.parquet"
-        query = f"SELECT * FROM public.{table_name}"
+        query = f"SELECT * FROM public_marts.marts_inclusion__{table_name}"
 
         logger.info("downloading data from query='%s'", query)
         df = pg_hook.get_pandas_df(sql=query)

pipeline/dags/main.py

@@ -89,7 +89,7 @@ def _geocode():
 
     dbt_seed = dbt_operator_factory(
         task_id="dbt_seed",
-        command="seed --full-refresh",
+        command="seed",
     )
 
     dbt_create_udfs = dbt_operator_factory(

pipeline/dbt/dbt_project.yml

@@ -28,18 +28,27 @@ models:
       +materialized: table
 
     marts:
-      api:
+      +schema: marts
+      +contract:
+        enforced: true
+
+      inclusion:
         +materialized: table
-      
+
       opendata:
         +schema: opendata
         +materialized: view
+        +contract:
+          enforced: false
 
 seeds:
   data_inclusion:
     schema:
       +schema: schema
+      +full_refresh: true
       +column_types:
         value: text
         label: text
-        description: text
+        description: text
+      +post-hook:
+        - "ALTER TABLE {{ this }} ADD UNIQUE (value)"

pipeline/dbt/macros/marts_inclusion__unnest_jsonb_list_to_table.sql

@@ -0,0 +1,18 @@
+{% macro unnest_jsonb_list_to_table(resource_type, nested_column) %}
+
+WITH {{ resource_type }}s AS (
+    SELECT * FROM {{ ref('int__union_{}s__enhanced'.format(resource_type) ) }}
+),
+
+final AS (
+    SELECT
+        _di_surrogate_id AS "{{ resource_type }}_surrogate_id",
+        value_           AS "value"
+    FROM
+        {{ resource_type }}s,
+        UNNEST({{ nested_column }}) AS value_
+)
+
+SELECT * FROM final
+
+{% endmacro %}