Add Source Freshness Results

README.md

@@ -37,6 +37,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.
@@ -75,6 +76,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/models/sources.yml

@@ -4,10 +4,6 @@ sources:
   - name: dummy_source
     database: "{% if target.type not in ('spark', 'databricks') %}{{ var('dbt_artifacts_database', target.database) }}{% endif %}"
     schema: "{{ target.schema }}"
-    freshness:
-      error_after: {count: 24, period: hour}
-      filter: dayname(updatedat) not in ('Sunday', 'Monday')
-    loaded_at_field: convert_timezone('UTC', load_timestamp)
     tables:
       - name: doesnt_exist
       - name: '"GROUP"'
@@ -16,7 +12,8 @@ sources:
     database: "{% if target.type not in ('spark', 'databricks') %}{{ var('dbt_artifacts_database', target.database) }}{% endif %}"
     schema: "{{ target.schema }}"
     freshness:
-      error_after: {count: 1, period: hour}
+      error_after: {count: 3, period: day}
+      warn_after: {count: 1, period: day}
     loaded_at_field: load_timestamp
     tables:
       - name: freshness

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:

integration_test_project/seeds/freshness.csv

@@ -3,3 +3,4 @@ id,load_timestamp
 102,2025-04-09 12:05:00
 103,2025-04-09 12:10:00
 104,2025-04-09 12:15:00
+105,2025-05-25 12:30:00

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 #}
+                {% if source.node.freshness.warn_after.count is not none %}{{ source.node.freshness.warn_after.count }}{% else %}null{% endif %}, {# warn_after_count #}
+                {% if source.node.freshness.warn_after.period is not none %}'{{ source.node.freshness.warn_after.period }}'{% else %}null{% endif %}, {# warn_after_period #}
+                {% if source.node.freshness.error_after.count is not none %}{{ source.node.freshness.error_after.count }}{% else %}null{% endif %}, {# error_after_count #}
+                {% if source.node.freshness.error_after.period is not none %}'{{ source.node.freshness.error_after.period }}'{% else %}null{% endif %}, {# 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 #}
+                {% if source.node.freshness.warn_after.count is not none %}{{ source.node.freshness.warn_after.count }}{% else %}null{% endif %}, {# warn_after_count #}
+                {% if source.node.freshness.warn_after.period is not none %}'{{ source.node.freshness.warn_after.period }}'{% else %}null{% endif %}, {# warn_after_period #}
+                {% if source.node.freshness.error_after.count is not none %}{{ source.node.freshness.error_after.count }}{% else %}null{% endif %}, {# error_after_count #}
+                {% if source.node.freshness.error_after.period is not none %}'{{ source.node.freshness.error_after.period }}'{% else %}null{% endif %}, {# 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

@@ -227,6 +227,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 %}