Capturing exposures data (#18) (#21)

* Capturing exposures data (#18)

* Added three models to capture exposures data: stg, dim and fct.

* Refactored fct model to adhere to dbt style guide.

* Fix indentation bug.

* CI Build

* CI Build

* CI Build

* Still fighting CI.

* Verify dbt is running from the correct dir.

Co-authored-by: Konrad <77880496+kon-mal@users.noreply.github.com>

Author: kgpayne

.circleci/config.yml

@@ -28,8 +28,8 @@ jobs:
           name: "Run Tests - Snowflake"
           command: |
             . venv/bin/activate
-            echo `pwd`
             cd integration_tests
+            echo `pwd`
             dbt --warn-error deps --target snowflake
             dbt --warn-error run-operation create_artifact_resources --target snowflake
             dbt --warn-error seed --target snowflake --full-refresh
@@ -43,7 +43,6 @@ jobs:
             dbt --warn-error docs generate --target snowflake
             dbt --warn-error run-operation upload_dbt_artifacts --args '{filenames: [catalog]}' --target snowflake
 
-
       - save_cache:
           key: deps1-{{ .Branch }}
           paths:

integration_tests/ci/sample.profiles.yml

@@ -6,7 +6,7 @@ config:
     send_anonymous_usage_stats: False
     use_colors: True
 
-integration_tests:
+dbt_artifacts_integration_tests:
   target: snowflake
   outputs:
     snowflake:

integration_tests/dbt_project.yml

@@ -2,7 +2,7 @@ name: 'artifacts_integration_tests'
 version: '1.0'
 config-version: 2
 
-profile: 'integration_tests'
+profile: 'dbt_artifacts_integration_tests'
 
 source-paths: ["models"]
 analysis-paths: ["analysis"]
@@ -12,8 +12,8 @@ macro-paths: ["macros"]
 
 target-path: "target"
 clean-targets:
-    - "target"
-    - "dbt_modules"
+- "target"
+- "dbt_modules"
 
 seeds:
   +quote_columns: false

models/fct_dbt__exposures_updates.sql

@@ -0,0 +1,57 @@
+with model_executions as (
+
+    select * from {{ ref('fct_dbt__model_executions') }}
+
+),
+
+exposures_record as (
+
+    select * from {{ ref('dim_dbt__exposures') }}
+
+)
+
+model_updates as (
+
+    select 
+        max(query_completed_at) as latest_update,
+        node_id
+    from model_executions
+    group by node_id
+
+),
+
+exposures_latest as (
+
+    select
+        artifact_generated_at as latest_generation,
+        node_id,
+        name,
+        type,
+        owner,
+        maturity,
+        package_name,
+        output_feeds
+    from exposures_record
+    where artifact_generated_at = (select max(artifact_generated_at) from exposures_record)
+
+),
+
+exposures_updates as (
+
+    select 
+        e.latest_generation,
+        e.node_id,
+        e.name, 
+        e.type,
+        e.owner,
+        e.maturity,
+        e.package_name,
+        e.output_feeds,
+        latest_update as feed_latest_update
+    from exposures e
+    left join model_updates m
+    on m.node_id = e.output_feeds
+
+)
+
+select * from exposures_updates

models/incremental/dim_dbt__exposures.sql

@@ -0,0 +1,44 @@
+{{ 
+  config(
+    materialized='incremental', 
+    unique_key='manifest_model_id'
+    ) 
+}}
+
+with dbt_models as (
+
+    select * from {{ ref('stg_dbt__exposures') }}
+
+),
+
+dbt_models_incremental as (
+
+    select *
+    from dbt_models
+
+    {% if is_incremental() %}
+    -- this filter will only be applied on an incremental run
+    where artifact_generated_at > (select max(artifact_generated_at) from {{ this }})
+    {% endif %}
+
+),
+
+fields as (
+
+     select 
+        t.manifest_model_id,
+        t.command_invocation_id,
+        t.artifact_generated_at,
+        t.node_id,
+        t.name,
+        t.type,
+        t.owner,
+        t.maturity,
+        f.value::string as output_feeds,
+        t.package_name
+    from dbt_models_incremental t,
+    lateral flatten(input => depends_on_nodes) f
+
+)
+
+select * from fields

models/staging/stg_dbt__exposures.sql

@@ -0,0 +1,52 @@
+with base as (
+
+    select *
+    from {{ ref('stg_dbt__artifacts') }}
+
+),
+
+manifests as (
+
+    select *
+    from base
+    where artifact_type = 'manifest.json'
+
+),
+
+flatten as (
+
+    select
+        command_invocation_id,
+        generated_at as artifact_generated_at,
+        node.key as node_id,
+        node.value:name::string as name,
+        to_array(node.value:depends_on:nodes) as depends_on_nodes,
+        to_array(node.value:sources:nodes) as depends_on_sources,
+        node.value:type::string as type,
+        node.value:owner:name::string as owner,
+        node.value:maturity::string as maturity,
+        node.value:package_name::string as package_name
+    from manifests,
+    lateral flatten(input => data:exposures) as node
+
+),
+
+surrogate_key as (
+
+    select
+        {{ dbt_utils.surrogate_key(['command_invocation_id', 'node_id']) }} as manifest_model_id,
+        command_invocation_id,
+        artifact_generated_at,
+        node_id,
+        name,
+        depends_on_nodes,
+        depends_on_sources,
+        type,
+        owner,
+        maturity,
+        package_name
+    from flatten
+
+)
+
+select * from surrogate_key