feat: [SNOW-1890085] check if dbt project exists before deploying

Author: sfc-gh-jwilkowski

src/snowflake/cli/_plugins/dbt/commands.py

@@ -36,7 +36,7 @@
 log = logging.getLogger(__name__)
 
 
-DBTNameArgument = identifier_argument(sf_object="DBT Object", example="my_pipeline")
+DBTNameArgument = identifier_argument(sf_object="DBT Project", example="my_pipeline")
 
 
 @app.command(

src/snowflake/cli/_plugins/dbt/manager.py

@@ -18,9 +18,10 @@
 
 import yaml
 from click import ClickException
+from snowflake.cli._plugins.object.manager import ObjectManager
 from snowflake.cli._plugins.stage.manager import StageManager
 from snowflake.cli.api.console import cli_console
-from snowflake.cli.api.constants import DEFAULT_SIZE_LIMIT_MB
+from snowflake.cli.api.constants import DEFAULT_SIZE_LIMIT_MB, ObjectType
 from snowflake.cli.api.identifiers import FQN
 from snowflake.cli.api.secure_path import SecurePath
 from snowflake.cli.api.sql_execution import SqlExecutionMixin
@@ -32,6 +33,12 @@ def list(self) -> SnowflakeCursor:  # noqa: A003
         query = "SHOW DBT PROJECTS"
         return self.execute_query(query)
 
+    @staticmethod
+    def exists(name: FQN) -> bool:
+        return ObjectManager().object_exists(
+            object_type=ObjectType.DBT_PROJECT.value.cli_name, fqn=name
+        )
+
     def deploy(
         self,
         path: SecurePath,
@@ -55,6 +62,11 @@ def deploy(
                         f"dbt-version was not provided and is not available in dbt_project.yml"
                     )
 
+        if self.exists(name=name) and force is not True:
+            raise ClickException(
+                f"DBT project {name} already exists. Use --force flag to overwrite"
+            )
+
         with cli_console.phase("Creating temporary stage"):
             stage_manager = StageManager()
             stage_fqn = FQN.from_string(f"dbt_{name}_stage").using_context()

src/snowflake/cli/api/constants.py

@@ -35,6 +35,7 @@ def __str__(self):
 
 class ObjectType(Enum):
     COMPUTE_POOL = ObjectNames("compute-pool", "compute pool", "compute pools")
+    DBT_PROJECT = ObjectNames("dbt-project", "DBT project", "DBT projects")
     DATABASE = ObjectNames("database", "database", "databases")
     FUNCTION = ObjectNames("function", "function", "functions")
     INTEGRATION = ObjectNames("integration", "integration", "integrations")
@@ -79,6 +80,7 @@ def __str__(self):
     ObjectType.APPLICATION.value.cli_name,
     ObjectType.APPLICATION_PACKAGE.value.cli_name,
     ObjectType.PROJECT.value.cli_name,
+    ObjectType.DBT_PROJECT.value.cli_name,
 }
 SUPPORTED_OBJECTS = sorted(OBJECT_TO_NAMES.keys() - UNSUPPORTED_OBJECTS)
 

tests/dbt/test_dbt_commands.py

@@ -55,10 +55,23 @@ def mock_cli_console(self):
         with mock.patch("snowflake.cli.api.console") as _fixture:
             yield _fixture
 
+    @pytest.fixture
+    def mock_exists(self):
+        with mock.patch(
+            "snowflake.cli._plugins.dbt.manager.DBTManager.exists", return_value=False
+        ) as _fixture:
+            yield _fixture
+
     @mock.patch("snowflake.cli._plugins.dbt.manager.StageManager.put_recursive")
     @mock.patch("snowflake.cli._plugins.dbt.manager.StageManager.create")
     def test_deploys_project_from_source(
-        self, mock_create, mock_put_recursive, mock_connect, runner, dbt_project_path
+        self,
+        mock_create,
+        mock_put_recursive,
+        mock_connect,
+        runner,
+        dbt_project_path,
+        mock_exists,
     ):
 
         result = runner.invoke(
@@ -88,7 +101,13 @@ def test_deploys_project_from_source(
     @mock.patch("snowflake.cli._plugins.dbt.manager.StageManager.put_recursive")
     @mock.patch("snowflake.cli._plugins.dbt.manager.StageManager.create")
     def test_dbt_version_from_option_has_precedence_over_file(
-        self, _mock_create, _mock_put_recursive, mock_connect, runner, dbt_project_path
+        self,
+        _mock_create,
+        _mock_put_recursive,
+        mock_connect,
+        runner,
+        dbt_project_path,
+        mock_exists,
     ):
         result = runner.invoke(
             [
@@ -110,11 +129,21 @@ def test_dbt_version_from_option_has_precedence_over_file(
 DBT_ADAPTER_VERSION='3.4.5'"""
         )
 
+    @pytest.mark.parametrize("exists", (True, False))
     @mock.patch("snowflake.cli._plugins.dbt.manager.StageManager.put_recursive")
     @mock.patch("snowflake.cli._plugins.dbt.manager.StageManager.create")
     def test_force_flag_uses_create_or_replace(
-        self, _mock_create, _mock_put_recursive, mock_connect, runner, dbt_project_path
+        self,
+        _mock_create,
+        _mock_put_recursive,
+        exists,
+        mock_connect,
+        runner,
+        dbt_project_path,
+        mock_exists,
     ):
+        mock_exists.return_value = exists
+
         result = runner.invoke(
             [
                 "dbt",
@@ -134,7 +163,13 @@ def test_force_flag_uses_create_or_replace(
     @mock.patch("snowflake.cli._plugins.dbt.manager.StageManager.put_recursive")
     @mock.patch("snowflake.cli._plugins.dbt.manager.StageManager.create")
     def test_execute_in_warehouse(
-        self, _mock_create, _mock_put_recursive, mock_connect, runner, dbt_project_path
+        self,
+        _mock_create,
+        _mock_put_recursive,
+        mock_connect,
+        runner,
+        dbt_project_path,
+        mock_exists,
     ):
 
         result = runner.invoke(
@@ -201,6 +236,27 @@ def test_raises_when_dbt_project_version_is_not_specified(
         )
         assert mock_connect.mocked_ctx.get_query() == ""
 
+    def test_raises_when_dbt_project_exists_and_is_not_force(
+        self, dbt_project_path, mock_connect, runner, mock_exists
+    ):
+        mock_exists.return_value = True
+
+        result = runner.invoke(
+            [
+                "dbt",
+                "deploy",
+                "TEST_PIPELINE",
+                f"--source={dbt_project_path}",
+            ],
+        )
+
+        assert result.exit_code == 1, result.output
+        assert (
+            "DBT project TEST_PIPELINE already exists. Use --force flag to overwrite"
+            in result.output
+        )
+        assert mock_connect.mocked_ctx.get_query() == ""
+
 
 class TestDBTExecute:
     @pytest.mark.parametrize(