Move pure logic into Python.

dbt/adapters/base/impl.py

@@ -1787,6 +1787,29 @@ def _get_adapter_specific_run_info(cls, config) -> Dict[str, Any]:
         """
         return {}
 
+    @available.parse_none
+    @classmethod
+    def get_hard_deletes_behavior(cls, config):
+        """Check the hard_deletes config enum, and the legacy invalidate_hard_deletes
+        config flag in order to determine which behavior should be used for deleted
+        records in a snapshot. The default is to ignore them."""
+        invalidate_hard_deletes = config.get("invalidate_hard_deletes", None)
+        hard_deletes = config.get("hard_deletes", None)
+
+        if invalidate_hard_deletes is not None and hard_deletes is not None:
+            raise DbtValidationError(
+                "You cannot set both the invalidate_hard_deletes and hard_deletes config properties on the same snapshot."
+            )
+
+        if invalidate_hard_deletes or hard_deletes == "invalidate":
+            return "invalidate"
+        elif hard_deletes == "new_record":
+            return "new_record"
+        elif hard_deletes is None or hard_deletes == "ignore":
+            return "ignore"
+
+        raise DbtValidationError("Invalid setting for property hard_deletes.")
+
 
 COLUMNS_EQUAL_SQL = """
 with diff_count as (

dbt/include/global_project/macros/materializations/snapshots/helpers.sql

@@ -38,28 +38,6 @@
     {{ return({'dbt_valid_to': 'dbt_valid_to', 'dbt_valid_from': 'dbt_valid_from', 'dbt_scd_id': 'dbt_scd_id', 'dbt_updated_at': 'dbt_updated_at', 'dbt_is_deleted': 'dbt_is_deleted'}) }}
 {% endmacro %}
 
-{# Check the hard_deletes config enum, and the legacy invalidate_hard_deletes
-   config flag in order to determine which behavior should be used for deleted
-   records in the current snapshot. The default is to ignore them. #}
-{% macro get_hard_delete_behavior() %}
-    {% set invalidate_hard_deletes = config.get('invalidate_hard_deletes') %}
-    {% set hard_deletes = config.get('hard_deletes') %}
-
-    {%  if invalidate_hard_deletes is not none and hard_deletes is not none %}
-        {% do exceptions.raise_compiler_error("You cannot set both the invalidate_hard_deletes and hard_deletes config properties on the same snapshot.") %}
-    {% endif %}
-
-    {%  if invalidate_hard_deletes or hard_deletes == 'invalidate' %}
-        {{ return('invalidate') }}
-    {% elif hard_deletes == 'new_record' %}
-        {{ return('new_record') }}
-    {% elif hard_deletes is none or hard_deletes == 'ignore' %}
-        {{ return('ignore') }}
-    {% else %}
-        {% do exceptions.raise_compiler_error("Invalid setting for property hard_deletes.") %}
-    {% endif %}
-{% endmacro %}
-
 {% macro default__snapshot_staging_table(strategy, source_sql, target_relation) -%}
     {% set columns = config.get('snapshot_table_column_names') or get_snapshot_table_column_names() %}
 

dbt/include/global_project/macros/materializations/snapshots/strategies.sql

@@ -54,7 +54,7 @@
     {# The model_config parameter is no longer used, but is passed in anyway for compatibility. #}
     {% set primary_key = config.get('unique_key') %}
     {% set updated_at = config.get('updated_at') %}
-    {% set hard_deletes = get_hard_delete_behavior() %}
+    {% set hard_deletes = adapter.get_hard_deletes_behavior(config) %}
     {% set invalidate_hard_deletes = hard_deletes == 'invalidate' %}
     {% set columns = config.get("snapshot_table_column_names") or get_snapshot_table_column_names() %}
 
@@ -143,7 +143,7 @@
     {# The model_config parameter is no longer used, but is passed in anyway for compatibility. #}
     {% set check_cols_config = config.get('check_cols') %}
     {% set primary_key = config.get('unique_key') %}
-    {% set hard_deletes = get_hard_delete_behavior() %}
+    {% set hard_deletes = adapter.get_hard_deletes_behavior(config) %}
     {% set invalidate_hard_deletes = hard_deletes == 'invalidate' %}
     {% set updated_at = config.get('updated_at') or snapshot_get_time() %}