classmethod scd_args on BaseRelation

dbt/adapters/base/relation.py

@@ -6,6 +6,7 @@
     Dict,
     FrozenSet,
     Iterator,
+    List,
     Optional,
     Set,
     Tuple,
@@ -341,6 +342,16 @@ def create(
         )
         return cls.from_dict(kwargs)
 
+    @classmethod
+    def scd_args(cls: Type[Self], primary_key: Union[str, List[str]], updated_at) -> List[str]:
+        scd_args = []
+        if isinstance(primary_key, list):
+            scd_args.extend(primary_key)
+        else:
+            scd_args.append(primary_key)
+        scd_args.append(updated_at)
+        return scd_args
+
     @property
     def can_be_renamed(self) -> bool:
         return self.type in self.renameable_relations

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

@@ -70,16 +70,8 @@
         ({{ snapshotted_rel }}.{{ columns.dbt_valid_from }} < {{ current_rel }}.{{ updated_at }})
     {%- endset %}
 
-    {% if primary_key | is_list %}
-        {% set scd_args = [] %}
-        {% for key in primary_key %}
-            {{ scd_args.append(key) }}
-        {% endfor %}
-        {{ scd_args.append(updated_at) }}
-        {% set scd_id_expr = snapshot_hash_arguments(scd_args) %}
-    {% else %}
-        {% set scd_id_expr = snapshot_hash_arguments([primary_key, updated_at]) %}
-    {% endif %}
+    {% set scd_args = api.Relation.scd_args(primary_key, updated_at) %}
+    {% set scd_id_expr = snapshot_hash_arguments([primary_key, updated_at]) %}
 
     {% do return({
         "unique_key": primary_key,