From ec0e01caf6a9a8fbe51571eee88f0c53546795f2 Mon Sep 17 00:00:00 2001
From: Charlie Bini <5003326+cbini@users.noreply.github.com>
Date: Tue, 28 Jan 2025 17:00:22 +0000
Subject: [PATCH] feat: dbt asset ignore conditions

---
 src/dbt/kipptaf/models/people/properties.yml       | 14 ++++++++++++++
 .../libraries/dbt/dagster_dbt_translator.py        |  9 +++++++--
 2 files changed, 21 insertions(+), 2 deletions(-)

diff --git a/src/dbt/kipptaf/models/people/properties.yml b/src/dbt/kipptaf/models/people/properties.yml
index 8a0daedaff..b8f29e6319 100644
--- a/src/dbt/kipptaf/models/people/properties.yml
+++ b/src/dbt/kipptaf/models/people/properties.yml
@@ -7,3 +7,17 @@ models:
         data_tests:
           - unique
           - not_null
+  - name: base_people__staff_roster_history
+    meta:
+      dagster:
+        automation_condition:
+          ignore:
+            keys:
+              - kipptaf/surveys/int_surveys__staff_information_survey_pivot
+  - name: int_people__staff_roster_history
+    meta:
+      dagster:
+        automation_condition:
+          ignore:
+            keys:
+              - kipptaf/surveys/int_surveys__staff_information_survey_pivot
diff --git a/src/teamster/libraries/dbt/dagster_dbt_translator.py b/src/teamster/libraries/dbt/dagster_dbt_translator.py
index 138d51e3a4..99826c15b3 100644
--- a/src/teamster/libraries/dbt/dagster_dbt_translator.py
+++ b/src/teamster/libraries/dbt/dagster_dbt_translator.py
@@ -1,6 +1,6 @@
 from typing import Any, Mapping
 
-from dagster import AssetKey, AutomationCondition
+from dagster import AssetKey, AssetSelection, AutomationCondition
 from dagster_dbt import DagsterDbtTranslator, DagsterDbtTranslatorSettings
 
 
@@ -46,7 +46,12 @@ def get_automation_condition(
             )
         else:
             return (
-                AutomationCondition.eager() | AutomationCondition.code_version_changed()
+                AutomationCondition.eager().ignore(
+                    AssetSelection.keys(
+                        *automation_condition_config.get("ignore", {}).get("keys", {})
+                    )
+                )
+                | AutomationCondition.code_version_changed()
             )
 
     def get_group_name(self, dbt_resource_props: Mapping[str, Any]) -> str | None: