Add Source Freshness Results

README.md

@@ -35,6 +35,7 @@ fct_dbt__model_executions
 fct_dbt__seed_executions
 fct_dbt__snapshot_executions
 fct_dbt__test_executions
+fct_dbt__source_executions
 ```
 
 See the generated [dbt docs site](https://brooklyn-data.github.io/dbt_artifacts/#!/overview) for documentation on each model.
@@ -73,6 +74,13 @@ packages:
     dbt run --select dbt_artifacts
     ```
 
+5. If you want to include results from the `dbt source freshness` command, you'll need to add the following flag in your `dbt_project.yml`.  This was recently added and made backwards-compatible to older core versions, so it's something that you have to opt into.
+
+    ```yml
+    flags:
+      source_freshness_run_project_hooks: true
+    ```
+
 ### Notes on upgrading
 
 Due to the structure of the project, when additional fields are added, the package needs to be re-run to ensure the tables include the new field, or it will simply error on the hook. These changes will always be implemented within a new **minor** version, so make sure that the version you use in `packages.yml` reflects this.

integration_test_project/dbt_project.yml

@@ -30,5 +30,10 @@ models:
 seeds:
   +quote_columns: false
 
+flags:
+  source_freshness_run_project_hooks: true
+  send_anonymous_usage_stats: false
+  use_colors: true
+
 on-run-end:
   - "{{ dbt_artifacts.upload_results(results) }}"

integration_test_project/profiles.yml

@@ -1,10 +1,6 @@
 # HEY! This file is used in the dbt-artifacts integrations tests with GitHub Actions.
 # You should __NEVER__ check credentials into version control. Thanks for reading :)
 
-config:
-  send_anonymous_usage_stats: False
-  use_colors: True
-
 dbt_artifacts:
   target: snowflake
   outputs:

macros/upload_individual_datasets/upload_source_executions.sql

@@ -0,0 +1,209 @@
+{% macro upload_source_executions(sources) %}
+    {{ return(adapter.dispatch('get_source_executions_dml_sql', 'dbt_artifacts')(sources)) }}
+{% endmacro %}
+
+{% macro default__get_source_executions_dml_sql(sources) -%}
+    {% if sources != [] %}
+
+        {% set source_execution_values %}
+        select
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(1) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(2) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(3) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(4) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(5) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(6) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(7) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(8) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(9) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(10) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(11) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(12) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(13) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(14) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(15) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(16) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(17) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(18) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(19) }},
+            {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(20)) }}
+        from values
+        {% for source in sources -%}
+            (
+                '{{ invocation_id }}', {# command_invocation_id #}
+                '{{ source.node.unique_id }}', {# node_id #}
+                '{{ run_started_at }}', {# run_started_at #}
+                '{{ source.thread_id }}', {# thread_id #}
+                '{{ source.status }}', {# status #}
+
+                {% set compile_started_at = (source.timing | selectattr("name", "eq", "compile") | first | default({}))["started_at"] %}
+                {% if compile_started_at %}'{{ compile_started_at }}'{% else %}null{% endif %}, {# compile_started_at #}
+                {% set query_completed_at = (source.timing | selectattr("name", "eq", "execute") | first | default({}))["completed_at"] %}
+                {% if query_completed_at %}'{{ query_completed_at }}'{% else %}null{% endif %}, {# query_completed_at #}
+
+                {{ source.execution_time }}, {# total_node_runtime #}
+                '{{ source.node.schema }}', {# schema #}
+                '{{ source.node.name }}', {# name #}
+                '{{ source.node.source_name }}', {# source_name #}
+                '{{ source.node.loaded_at_field }}', {# loaded_at_field #}
+                {{ source.node.freshness.warn_after.count }}, {# warn_after_count #}
+                '{{ source.node.freshness.warn_after.period }}', {# warn_after_period #}
+                {{ source.node.freshness.error_after.count }}, {# error_after_count #}
+                '{{ source.node.freshness.error_after.period }}', {# error_after_period #}
+                '{{ source.max_loaded_at }}', {# max_loaded_at #}
+                '{{ source.snapshotted_at }}', {# snapshotted_at #}
+                {{ source.age }}, {# age #}
+                '{{ tojson(source.adapter_response) | replace("\\", "\\\\") | replace("'", "\\'") | replace('"', '\\"') }}' {# adapter_response #}
+            )
+            {%- if not loop.last %},{%- endif %}
+        {% endfor %}
+        {% endset %}
+        {{ source_execution_values }}
+    {% else %}
+        {{ return ("") }}
+    {% endif %}
+{% endmacro -%}
+
+
+{% macro bigquery__get_source_executions_dml_sql(sources) -%}
+    {% if sources != [] %}
+
+        {% set source_execution_values %}
+        {% for source in sources -%}
+            (
+                '{{ invocation_id }}', {# command_invocation_id #}
+                '{{ source.node.unique_id }}', {# node_id #}
+                '{{ run_started_at }}', {# run_started_at #}
+                '{{ source.thread_id }}', {# thread_id #}
+                '{{ source.status }}', {# status #}
+
+                {% set compile_started_at = (source.timing | selectattr("name", "eq", "compile") | first | default({}))["started_at"] %}
+                {% if compile_started_at %}'{{ compile_started_at }}'{% else %}null{% endif %}, {# compile_started_at #}
+                {% set query_completed_at = (source.timing | selectattr("name", "eq", "execute") | first | default({}))["completed_at"] %}
+                {% if query_completed_at %}'{{ query_completed_at }}'{% else %}null{% endif %}, {# query_completed_at #}
+
+                {{ source.execution_time }}, {# total_node_runtime #}
+                '{{ source.node.schema }}', {# schema #}
+                '{{ source.node.name }}', {# name #}
+                '{{ source.node.source_name }}', {# source_name #}
+                '{{ source.node.loaded_at_field }}', {# loaded_at_field #}
+                '{{ source.node.freshness.warn_after.count }}', {# warn_after_count #}
+                '{{ source.node.freshness.warn_after.period }}', {# warn_after_period #}
+                '{{ source.node.freshness.error_after.count }}', {# error_after_count #}
+                '{{ source.node.freshness.error_after.period }}', {# error_after_period #}
+                '{{ source.max_loaded_at }}', {# max_loaded_at #}
+                '{{ source.snapshotted_at }}', {# snapshotted_at #}
+                {{ source.age }}, {# age #}
+                {{ adapter.dispatch('parse_json', 'dbt_artifacts')(tojson(model.adapter_response) | replace("\\", "\\\\") | replace("'", "\\'") | replace('"', '\\"')) }} {# adapter_response #}
+            )
+            {%- if not loop.last %},{%- endif %}
+        {% endfor %}
+        {% endset %}
+        {{ source_execution_values }}
+    {% else %}
+        {{ return ("") }}
+    {% endif %}
+{% endmacro -%}
+
+
+{% macro snowflake__get_source_executions_dml_sql(sources) -%}
+    {% if sources != [] %}
+
+        {% set source_execution_values %}
+        select
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(1) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(2) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(3) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(4) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(5) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(6) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(7) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(8) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(9) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(10) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(11) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(12) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(13) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(14) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(15) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(16) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(17) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(18) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(19) }},
+            {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(20)) }}
+        from values
+        {% for source in sources -%}
+            (
+                '{{ invocation_id }}', {# command_invocation_id #}
+                '{{ source.node.unique_id }}', {# node_id #}
+                '{{ run_started_at }}', {# run_started_at #}
+                '{{ source.thread_id }}', {# thread_id #}
+                '{{ source.status }}', {# status #}
+
+                {% set compile_started_at = (source.timing | selectattr("name", "eq", "compile") | first | default({}))["started_at"] %}
+                {% if compile_started_at %}'{{ compile_started_at }}'{% else %}null{% endif %}, {# compile_started_at #}
+                {% set query_completed_at = (source.timing | selectattr("name", "eq", "execute") | first | default({}))["completed_at"] %}
+                {% if query_completed_at %}'{{ query_completed_at }}'{% else %}null{% endif %}, {# query_completed_at #}
+
+                {{ source.execution_time }}, {# total_node_runtime #}
+                '{{ source.node.schema }}', {# schema #}
+                '{{ source.node.name }}', {# name #}
+                '{{ source.node.source_name }}', {# source_name #}
+                '{{ source.node.loaded_at_field }}', {# loaded_at_field #}
+                {{ source.node.freshness.warn_after.count }}, {# warn_after_count #}
+                '{{ source.node.freshness.warn_after.period }}', {# warn_after_period #}
+                {{ source.node.freshness.error_after.count }}, {# error_after_count #}
+                '{{ source.node.freshness.error_after.period }}', {# error_after_period #}
+                '{{ source.max_loaded_at }}', {# max_loaded_at #}
+                '{{ source.snapshotted_at }}', {# snapshotted_at #}
+                {{ source.age }}, {# age #}
+                '{{ tojson(source.adapter_response) | replace("\\", "\\\\") | replace("'", "\\'") | replace('"', '\\"') }}' {# adapter_response #}
+            )
+            {%- if not loop.last %},{%- endif %}
+        {% endfor %}
+        {% endset %}
+        {{ source_execution_values }}
+    {% else %}
+        {{ return ("") }}
+    {% endif %}
+{% endmacro -%}
+
+{% macro postgres__get_source_executions_dml_sql(sources) -%}
+    {% if sources != [] %}
+
+        {% set source_execution_values %}
+        {% for source in sources -%}
+            (
+                '{{ invocation_id }}', {# command_invocation_id #}
+                '{{ source.node.unique_id }}', {# node_id #}
+                '{{ run_started_at }}', {# run_started_at #}
+                '{{ source.thread_id }}', {# thread_id #}
+                '{{ source.status }}', {# status #}
+
+                {% set compile_started_at = (source.timing | selectattr("name", "eq", "compile") | first | default({}))["started_at"] %}
+                {% if compile_started_at %}'{{ compile_started_at }}'{% else %}null{% endif %}, {# compile_started_at #}
+                {% set query_completed_at = (source.timing | selectattr("name", "eq", "execute") | first | default({}))["completed_at"] %}
+                {% if query_completed_at %}'{{ query_completed_at }}'{% else %}null{% endif %}, {# query_completed_at #}
+
+                {{ source.execution_time }}, {# total_node_runtime #}
+                '{{ source.node.schema }}', {# schema #}
+                '{{ source.node.name }}', {# name #}
+                '{{ source.node.source_name }}', {# source_name #}
+                '{{ source.node.loaded_at_field }}', {# loaded_at_field #}
+                '{{ source.node.freshness.warn_after.count }}', {# warn_after_count #}
+                '{{ source.node.freshness.warn_after.period }}', {# warn_after_period #}
+                '{{ source.node.freshness.error_after.count }}', {# error_after_count #}
+                '{{ source.node.freshness.error_after.period }}', {# error_after_period #}
+                '{{ source.max_loaded_at }}', {# max_loaded_at #}
+                '{{ source.snapshotted_at }}', {# snapshotted_at #}
+                {{ source.age }}, {# age #}
+                $${{ tojson(model.adapter_response) }}$$ {# adapter_response #}
+            )
+            {%- if not loop.last %},{%- endif %}
+        {% endfor %}
+        {% endset %}
+        {{ source_execution_values }}
+    {% else %}
+        {{ return ("") }}
+    {% endif %}
+{% endmacro -%}

macros/upload_results/get_column_name_lists.sql

@@ -221,6 +221,31 @@
             all_results
         )
 
+    {% elif dataset == 'source_executions' %}
+
+        (
+            command_invocation_id,
+            node_id,
+            run_started_at,
+            thread_id,
+            status,
+            compile_started_at,
+            query_completed_at,
+            total_node_runtime,
+            schema,
+            name,
+            source_name,
+            loaded_at_field,
+            warn_after_count,
+            warn_after_period,
+            error_after_count,
+            error_after_period,
+            max_loaded_at,
+            snapshotted_at,
+            age,
+            adapter_response
+        )
+
     {% else %}
 
         /* No column list available */

macros/upload_results/get_dataset_content.sql

@@ -1,6 +1,6 @@
 {% macro get_dataset_content(dataset) %}
 
-    {% if dataset in ['model_executions', 'seed_executions', 'test_executions', 'snapshot_executions'] %}
+    {% if dataset in ['model_executions', 'seed_executions', 'test_executions', 'snapshot_executions', 'source_executions'] %}
         {# Executions make use of the results object #}
         {% set objects = results | selectattr("node.resource_type", "equalto", dataset.split("_")[0]) | list %}
     {% elif dataset in ['seeds', 'snapshots', 'tests', 'models'] %}

macros/upload_results/get_table_content_values.sql

@@ -10,6 +10,8 @@
         {% set content = dbt_artifacts.upload_test_executions(objects_to_upload) %}
     {% elif dataset == 'snapshot_executions' %}
         {% set content = dbt_artifacts.upload_snapshot_executions(objects_to_upload) %}
+    {% elif dataset == 'source_executions' %}
+        {% set content = dbt_artifacts.upload_source_executions(objects_to_upload) %}
     {% elif dataset == 'exposures' %}
         {% set content = dbt_artifacts.upload_exposures(objects_to_upload) %}
     {% elif dataset == 'models' %}

macros/upload_results/upload_results.sql

@@ -7,7 +7,7 @@
         {% set datasets_to_load = ['exposures', 'seeds', 'snapshots', 'invocations', 'sources', 'tests', 'models'] %}
         {% if results != [] %}
             {# When executing, and results are available, then upload the results #}
-            {% set datasets_to_load = ['model_executions', 'seed_executions', 'test_executions', 'snapshot_executions'] + datasets_to_load %}
+            {% set datasets_to_load = ['model_executions', 'seed_executions', 'test_executions', 'snapshot_executions', 'source_executions'] + datasets_to_load %}
         {% endif %}
 
         {# Upload each data set in turn #}

models/docs.md

@@ -412,3 +412,45 @@ Response provided by the adapter as JSON.
 All results as a JSON blob
 
 {% enddocs %}
+
+{% docs warn_after_count %}
+
+Positive integer indicating a threshold to warn (used alongside warn_after_period)
+
+{% enddocs %}
+
+{% docs warn_after_period %}
+
+Used alongside warn_after_count to indicate number of periods (minutes, hours, days) for a warn threshold
+
+{% enddocs %}
+
+{% docs error_after_count %}
+
+Positive integer indicating a threshold to error (used alongside error_after_period)
+
+{% enddocs %}
+
+{% docs error_after_period %}
+
+Used alongside error_after_count to indicate number of periods (minutes, hours, days) for a error threshold
+
+{% enddocs %}
+
+{% docs max_loaded_at %}
+
+Max value of loaded_at_field timestamp in the source table when queried
+
+{% enddocs %}
+
+{% docs snapshotted_at %}
+
+Current timestamp when querying
+
+{% enddocs %}
+
+{% docs age %}
+
+Interval between max_loaded_at and snapshotted_at, calculated in python to handle timezone complexity
+
+{% enddocs %}

models/fct_dbt__source_executions.sql

@@ -0,0 +1,36 @@
+with base as (
+
+    select *
+    from {{ ref('stg_dbt__source_executions') }}
+
+),
+
+source_executions as (
+
+    select
+        source_execution_id,
+        command_invocation_id,
+        node_id,
+        run_started_at,
+        thread_id,
+        status,
+        compile_started_at,
+        query_completed_at,
+        total_node_runtime,
+        schema, -- noqa
+        name,
+        source_name,
+        loaded_at_field,
+        warn_after_count,
+        warn_after_period,
+        error_after_count,
+        error_after_period,
+        max_loaded_at,
+        snapshotted_at,
+        age,
+        adapter_response
+    from base
+
+)
+
+select * from source_executions