Merge pull request #492 from brooklyn-data/bug/missing-datetime-object-fix

Bug: Handle event time flags for microbatch

Author: mtcarlone

.gitignore

@@ -14,6 +14,10 @@ env.sh
 .python-version
 .vscode
 .venv
+.env
+.secrets
+*.lock
+**/*/package-lock.yml
 pyproject.toml
 
 .DS_Store

README.md

@@ -47,7 +47,7 @@ See the generated [dbt docs site](https://brooklyn-data.github.io/dbt_artifacts/
 ```
 packages:
   - package: brooklyn-data/dbt_artifacts
-    version: 2.9.1
+    version: 2.9.2
 ```
 
 :construction_worker: Make sure to fix at least the **minor** version, to avoid issues when a new release is open. See the notes on upgrading below for more detail.

dbt_project.yml

@@ -1,5 +1,5 @@
 name: "dbt_artifacts"
-version: "2.9.1"
+version: "2.9.2"
 config-version: 2
 require-dbt-version: [">=1.3.0", "<1.10.0"]
 profile: "dbt_artifacts"

integration_test_project/models/microbatch.sql

@@ -1,11 +1,16 @@
 {{
   config(
-    materialized = 'incremental',
-    incremental_strategy = 'microbatch',
-    unique_key = 'transaction_id',
-    event_time = 'transaction_ts',
-    begin = (modules.datetime.datetime.now().replace(hour=6, minute=0, second=0) - modules.datetime.timedelta(2)).isoformat(),
-    batch_size = 'day'
+        materialized = 'incremental',
+        incremental_strategy = 'microbatch',
+        event_time = 'transaction_ts',
+        begin = (modules.datetime.datetime.now().replace(hour=6, minute=0, second=0) - modules.datetime.timedelta(2)).isoformat(),
+        batch_size = 'day',
+
+        unique_key = 'transaction_id',
+        partition_by = {
+            'field': 'store_name',
+            'data_type': 'text',
+        }
     )
 }}
 

integration_test_project/models/sources.yml

@@ -11,3 +11,12 @@ sources:
     tables:
       - name: doesnt_exist
       - name: '"GROUP"'
+
+  - name: freshness_check
+    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}
+    loaded_at_field: load_timestamp
+    tables:
+      - name: freshness

integration_test_project/seeds/freshness.csv

@@ -0,0 +1,5 @@
+id,load_timestamp
+101,2025-04-09 12:00:00
+102,2025-04-09 12:05:00
+103,2025-04-09 12:10:00
+104,2025-04-09 12:15:00

integration_test_project/seeds/seeds.yml

@@ -0,0 +1,8 @@
+---
+
+seeds:
+  - name: freshness
+    config:
+      column_types:
+        id: integer
+        load_timestamp: timestamp

macros/upload_individual_datasets/upload_invocations.sql

@@ -17,6 +17,18 @@
         {% endif %}
     {% endif %}
 
+    {% if "event_time_start" in invocation_args_dict and invocation_args_dict.strftime is not none %}
+        {% do invocation_args_dict.update(
+            {"event_time_start": invocation_args_dict.event_time_start.strftime(dbt_artifacts.get_strftime_format())}
+        ) %}
+    {% endif %}
+    {% if "event_time_end" in invocation_args_dict and invocation_args_dict.strftime is not none %}
+        {% do invocation_args_dict.update(
+            {"event_time_end": invocation_args_dict.event_time_end.strftime(dbt_artifacts.get_strftime_format())}
+        ) %}
+    {% endif %}
+
+    {{ log(invocation_args_dict) }}
     {{ return(adapter.dispatch("get_invocations_dml_sql", "dbt_artifacts")()) }}
 {%- endmacro %}
 

macros/utils/copy_model.sql

@@ -2,7 +2,7 @@
     {% set model_copy = model.copy() %}
     {% do model_copy.pop('raw_code', None) %}
     {% if model_copy.config.begin %}
-        {% set _ = model_copy.config.update({"begin": model_copy.config.begin.strftime("%Y-%m-%dT%H:%M:%S.%f")}) %}
+        {% set _ = model_copy.config.update({"begin": model_copy.config.begin.strftime(dbt_artifacts.get_strftime_format())}) %}
     {% endif %}
 
     {{ return(model_copy) }}

macros/utils/get_strftime_fmt.sql

@@ -0,0 +1,3 @@
+{% macro get_strftime_format() %}
+    {{ return("%Y-%m-%dT%H:%M:%S.%f") }}
+{% endmacro %}