feat: [SNOW-1890085] dbt deploy: add support for dbt-version flag

Author: sfc-gh-jwilkowski

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

@@ -66,6 +66,10 @@ def deploy_dbt(
         False,
         help="Overwrites conflicting files in the project, if any.",
     ),
+    dbt_version: Optional[str] = typer.Option(
+        None,
+        help="Version of dbt tool to be used. Taken from dbt_project.yml if not provided.",
+    ),
     **options,
 ) -> CommandResult:
     """
@@ -76,7 +80,9 @@ def deploy_dbt(
         path = Path.cwd()
     else:
         path = Path(source)
-    return QueryResult(DBTManager().deploy(path.resolve(), name, force=force))
+    return QueryResult(
+        DBTManager().deploy(path.resolve(), name, dbt_version, force=force)
+    )
 
 
 # `execute` is a pass through command group, meaning that all params after command should be passed over as they are,

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

@@ -15,7 +15,9 @@
 from __future__ import annotations
 
 from pathlib import Path
+from typing import Optional
 
+import yaml
 from click import ClickException
 from snowflake.cli._plugins.stage.manager import StageManager
 from snowflake.cli.api.console import cli_console
@@ -42,11 +44,23 @@ def list(self) -> SnowflakeCursor:  # noqa: A003
         query = "SHOW DBT PROJECT"
         return self.execute_query(query)
 
-    def deploy(self, path: Path, name: FQN, force: bool) -> SnowflakeCursor:
+    def deploy(
+        self, path: Path, name: FQN, dbt_version: Optional[str], force: bool
+    ) -> SnowflakeCursor:
         # TODO: what to do with force?
         if not path.joinpath("dbt_project.yml").exists():
             raise ClickException(f"dbt_project.yml does not exist in provided path.")
 
+        if dbt_version is None:
+            with path.joinpath("dbt_project.yml").open() as fd:
+                dbt_project_config = yaml.safe_load(fd)
+                try:
+                    dbt_version = dbt_project_config["version"]
+                except (KeyError, TypeError):
+                    raise ClickException(
+                        f"dbt-version was not provided and is not available in dbt_project.yml"
+                    )
+
         with cli_console.phase("Creating temporary stage"):
             stage_manager = StageManager()
             stage_fqn = FQN.from_string(f"dbt_{name}_stage").using_context()
@@ -58,7 +72,7 @@ def deploy(self, path: Path, name: FQN, force: bool) -> SnowflakeCursor:
             cli_console.step(f"Copied {len(results)} files")
 
         with cli_console.phase("Creating DBT project"):
-            query = f"CREATE OR REPLACE DBT PROJECT {name} FROM {stage_name}"
+            query = f"CREATE OR REPLACE DBT PROJECT {name} FROM {stage_name} DBT_VERSION='{dbt_version}'"
             return self.execute_query(query)
 
     def execute(self, dbt_command: str, name: str, *dbt_cli_args):

tests/dbt/test_dbt_commands.py

@@ -17,6 +17,7 @@
 from unittest import mock
 
 import pytest
+import yaml
 from snowflake.cli.api.identifiers import FQN
 
 
@@ -44,6 +45,8 @@ def dbt_project_path(self, tmp_path_factory):
         source_path = tmp_path_factory.mktemp("dbt_project")
         dbt_file = source_path / "dbt_project.yml"
         dbt_file.touch()
+        with dbt_file.open(mode="w") as fd:
+            yaml.dump({"version": "1.2.3"}, fd)
         yield source_path
 
     @pytest.fixture
@@ -64,14 +67,35 @@ def test_deploys_project_from_source(
         assert result.exit_code == 0, result.output
         assert (
             mock_connect.mocked_ctx.get_query()
-            == "CREATE OR REPLACE DBT PROJECT TEST_PIPELINE FROM @MockDatabase.MockSchema.dbt_TEST_PIPELINE_stage"
+            == "CREATE OR REPLACE DBT PROJECT TEST_PIPELINE FROM @MockDatabase.MockSchema.dbt_TEST_PIPELINE_stage DBT_VERSION='1.2.3'"
         )
         stage_fqn = FQN.from_string(f"dbt_TEST_PIPELINE_stage").using_context()
         mock_create.assert_called_once_with(stage_fqn, temporary=True)
         mock_put_recursive.assert_called_once_with(
             dbt_project_path, "@MockDatabase.MockSchema.dbt_TEST_PIPELINE_stage"
         )
 
+    @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
+    ):
+        result = runner.invoke(
+            [
+                "dbt",
+                "deploy",
+                "TEST_PIPELINE",
+                f"--source={dbt_project_path}",
+                "--dbt-version=2.3.4",
+            ]
+        )
+
+        assert result.exit_code == 0, result.output
+        assert (
+            mock_connect.mocked_ctx.get_query()
+            == "CREATE OR REPLACE DBT PROJECT TEST_PIPELINE FROM @MockDatabase.MockSchema.dbt_TEST_PIPELINE_stage DBT_VERSION='2.3.4'"
+        )
+
     def test_raises_when_dbt_project_is_not_available(
         self, dbt_project_path, mock_connect, runner
     ):
@@ -86,6 +110,24 @@ def test_raises_when_dbt_project_is_not_available(
         assert "dbt_project.yml does not exist in provided path." in result.output
         assert mock_connect.mocked_ctx.get_query() == ""
 
+    def test_raises_when_dbt_project_version_is_not_specified(
+        self, dbt_project_path, mock_connect, runner
+    ):
+        dbt_file = dbt_project_path / "dbt_project.yml"
+        with dbt_file.open(mode="w") as fd:
+            yaml.dump({}, fd)
+
+        result = runner.invoke(
+            ["dbt", "deploy", "TEST_PIPELINE", f"--source={dbt_project_path}"]
+        )
+
+        assert result.exit_code == 1, result.output
+        assert (
+            "dbt-version was not provided and is not available in dbt_project.yml"
+            in result.output
+        )
+        assert mock_connect.mocked_ctx.get_query() == ""
+
 
 class TestDBTExecute:
     @pytest.mark.parametrize(