Allow snapshots to be defined with YAML only.

.changes/unreleased/Features-20240920-110447.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Allow snapshots to be defined in YAML.
+time: 2024-09-20T11:04:47.703117-04:00
+custom:
+  Author: peterallenwebb
+  Issue: "10246"

core/dbt/parser/schemas.py

@@ -205,6 +205,7 @@
 
             # PatchParser.parse()
             if "snapshots" in dct:
+                self._add_yaml_snapshot_nodes_to_manifest(dct["snapshots"], block)
                 snapshot_parse_result = TestablePatchParser(self, yaml_block, "snapshots").parse()
                 for test_block in snapshot_parse_result.test_blocks:
                     self.generic_test_parser.parse_tests(test_block)
@@ -265,6 +266,55 @@
                 saved_query_parser = SavedQueryParser(self, yaml_block)
                 saved_query_parser.parse()
 
+    def _add_yaml_snapshot_nodes_to_manifest(
+        self, snapshots: List[Dict[str, Any]], block: FileBlock
+    ) -> None:
+        """We support the creation of simple snapshots in yaml, without an
+        accompanying SQL definition. For such snapshots, the user must supply
+        a 'relation' property to indicate the target of the snapshot. This
+        function looks for such snapshots and adds a node to manifest for each
+        one we find, since they were not added during SQL parsing."""
+
+        rebuild_refs = False
+        for snapshot in snapshots:
+            if "relation" in snapshot:
+                from dbt.parser import SnapshotParser
+
+                if "name" not in snapshot:
+                    raise ParsingError("A snapshot must define the 'name' property. ")
+
+                # Reuse the logic of SnapshotParser as far as possible to create
+                # a new node we can add to the manifest.
+                parser = SnapshotParser(self.project, self.manifest, self.root_project)
+                fqn = parser.get_fqn_prefix(block.path.relative_path)
+                fqn.append(snapshot["name"])
+                snapshot_node = parser._create_parsetime_node(
+                    block,
+                    self.get_compiled_path(block),
+                    parser.initial_config(fqn),
+                    fqn,
+                    snapshot["name"],
+                )
+
+                # Parse the expected ref() or source() expression given by
+                # 'relation' so that we know what we are snapshotting.
+                source_or_ref = statically_parse_ref_or_source(snapshot["relation"])
+                if isinstance(source_or_ref, RefArgs):
+                    snapshot_node.refs.append(source_or_ref)
+                else:
+                    snapshot_node.sources.append(source_or_ref)
+
+                # Implement the snapshot SQL as a simple select *
+                snapshot_node.raw_code = "select * from {{ " + snapshot["relation"] + " }}"
+
+                # Add our new node to the manifest, and note that ref lookup collections
+                # will need to be rebuilt.
+                self.manifest.add_node_nofile(snapshot_node)
+                rebuild_refs = True
+
+        if rebuild_refs:
+            self.manifest.rebuild_ref_lookup()
+
 
 Parsed = TypeVar("Parsed", UnpatchedSourceDefinition, ParsedNodePatch, ParsedMacroPatch)
 NodeTarget = TypeVar("NodeTarget", UnparsedNodeUpdate, UnparsedAnalysisUpdate, UnparsedModelUpdate)

tests/functional/snapshots/fixtures.py

@@ -291,6 +291,19 @@
 {% endsnapshot %}
 """
 
+snapshots_pg__snapshot_yml = """
+version: 2
+snapshots:
+  - name: snapshot_actual
+    relation: "ref('seed')"
+    config:
+      unique_key: "id || '-' || first_name"
+      strategy: timestamp
+      updated_at: updated_at
+      meta:
+        owner: 'a_owner'
+"""
+
 snapshots_pg__snapshot_no_target_schema_sql = """
 {% snapshot snapshot_actual %}
 

tests/functional/snapshots/test_basic_snapshot.py

@@ -20,6 +20,7 @@
     seeds__seed_newcol_csv,
     snapshots_pg__snapshot_no_target_schema_sql,
     snapshots_pg__snapshot_sql,
+    snapshots_pg__snapshot_yml,
     snapshots_pg_custom__snapshot_sql,
     snapshots_pg_custom_namespaced__snapshot_sql,
 )
@@ -372,3 +373,24 @@ def test_updated_at_snapshot(self, project):
 class TestRefUpdatedAtCheckCols(UpdatedAtCheckCols):
     def test_updated_at_ref(self, project):
         ref_setup(project, num_snapshot_models=2)
+
+
+class BasicYaml(Basic):
+    @pytest.fixture(scope="class")
+    def snapshots(self):
+        """Overrides the same function in Basic to use the YAML method of
+        defining a snapshot."""
+        return {"snapshot.yml": snapshots_pg__snapshot_yml}
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        """Overrides the same function in Basic to use a modified version of
+        schema.yml without snapshot config."""
+        return {
+            "ref_snapshot.sql": models__ref_snapshot_sql,
+        }
+
+
+class TestBasicSnapshotYaml(BasicYaml):
+    def test_basic_snapshot_yaml(self, project):
+        snapshot_setup(project, num_snapshot_models=1)