From bdf28d7eff13edcbe0d41746e50ffa8edcfa3cf0 Mon Sep 17 00:00:00 2001
From: Gerda Shank <gerda@dbtlabs.com>
Date: Fri, 1 Nov 2024 13:47:28 -0400
Subject: [PATCH] Support --empty option for 'snapshot' command (#10962)

---
 .../unreleased/Features-20241031-163149.yaml  |  6 +++
 core/dbt/cli/main.py                          |  1 +
 .../snapshots/test_snapshot_empty.py          | 40 +++++++++++++++++++
 3 files changed, 47 insertions(+)
 create mode 100644 .changes/unreleased/Features-20241031-163149.yaml
 create mode 100644 tests/functional/snapshots/test_snapshot_empty.py

diff --git a/.changes/unreleased/Features-20241031-163149.yaml b/.changes/unreleased/Features-20241031-163149.yaml
new file mode 100644
index 00000000000..209f2180daa
--- /dev/null
+++ b/.changes/unreleased/Features-20241031-163149.yaml
@@ -0,0 +1,6 @@
+kind: Features
+body: Support --empty for snapshots
+time: 2024-10-31T16:31:49.926164-04:00
+custom:
+  Author: gshank
+  Issue: "10372"
diff --git a/core/dbt/cli/main.py b/core/dbt/cli/main.py
index 610163042c4..a9de9441365 100644
--- a/core/dbt/cli/main.py
+++ b/core/dbt/cli/main.py
@@ -716,6 +716,7 @@ def seed(ctx, **kwargs):
 @cli.command("snapshot")
 @click.pass_context
 @global_flags
+@p.empty
 @p.exclude
 @p.profiles_dir
 @p.project_dir
diff --git a/tests/functional/snapshots/test_snapshot_empty.py b/tests/functional/snapshots/test_snapshot_empty.py
new file mode 100644
index 00000000000..a3f648a968f
--- /dev/null
+++ b/tests/functional/snapshots/test_snapshot_empty.py
@@ -0,0 +1,40 @@
+import pytest
+
+from dbt.tests.util import run_dbt
+
+my_model_sql = """
+select 1 as id, {{ dbt.current_timestamp() }} as updated_at
+"""
+
+snapshots_yml = """
+snapshots:
+  - name: my_snapshot
+    relation: "ref('my_model')"
+    config:
+      unique_key: id
+      strategy: check
+      check_cols: all
+      dbt_valid_to_current: "date('9999-12-31')"
+"""
+
+
+class TestSnapshotEmpty:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "my_model.sql": my_model_sql,
+            "snapshots.yml": snapshots_yml,
+        }
+
+    def test_check(self, project):
+        run_dbt(["run"])
+        run_dbt(["snapshot", "--empty"])
+
+        query = "select id, updated_at, dbt_valid_from, dbt_valid_to from {database}.{schema}.my_snapshot order by updated_at asc"
+        snapshot_out1 = project.run_sql(query, fetch="all")
+        assert snapshot_out1 == []
+
+        run_dbt(["run"])
+        run_dbt(["snapshot", "--empty"])
+        snapshot_out2 = project.run_sql(query, fetch="all")
+        assert snapshot_out2 == []