feat: [SNOW-1890085] add dbt execute command

Author: sfc-gh-jwilkowski

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

@@ -16,6 +16,7 @@
 
 import logging
 
+import typer
 from snowflake.cli._plugins.dbt.manager import DBTManager
 from snowflake.cli.api.commands.snow_typer import SnowTyperFactory
 from snowflake.cli.api.output.types import CommandResult, QueryResult
@@ -39,3 +40,28 @@ def list_dbts(
     List all dbt projects on Snowflake.
     """
     return QueryResult(DBTManager().list())
+
+
+@app.command(
+    "execute",
+    requires_connection=True,
+    context_settings={"allow_extra_args": True, "ignore_unknown_options": True},
+)
+def execute(
+    ctx: typer.Context,
+    dbt_command: str = typer.Argument(
+        help="dbt command to execute, i. e. run, compile, seed...",
+    ),
+    name: str = typer.Option(
+        default=...,
+        help="Name of the dbt object to execute command on.",
+    ),
+    **options,
+) -> CommandResult:
+    """
+    Execute command on dbt in Snowflake project.
+    """
+    # ctx.args are parameters that were not captured as known cli params (those are in **options).
+    # as a consequence, we don't support passing params known to snowflake cli further to dbt
+    dbt_cli_args = ctx.args
+    return QueryResult(DBTManager().execute(dbt_command, name, *dbt_cli_args))

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

@@ -20,5 +20,11 @@
 
 class DBTManager(SqlExecutionMixin):
     def list(self) -> SnowflakeCursor:  # noqa: A003
-        query = f"show dbt"
+        query = "SHOW DBT"
+        return self.execute_query(query)
+
+    def execute(self, dbt_command: str, name: str, *dbt_cli_args):
+        query = f"EXECUTE DBT {name} {dbt_command}"
+        if dbt_cli_args:
+            query += " " + " ".join([arg for arg in dbt_cli_args])
         return self.execute_query(query)

tests/dbt/test_dbt_commands.py

@@ -26,9 +26,52 @@ def mock_connect(mock_ctx):
         yield _fixture
 
 
-def test_list_dbt_objects(mock_connect, runner):
+def test_dbt_list(mock_connect, runner):
 
     result = runner.invoke(["dbt", "list"])
 
     assert result.exit_code == 0, result.output
-    assert mock_connect.mocked_ctx.get_query() == "show dbt"
+    assert mock_connect.mocked_ctx.get_query() == "SHOW DBT"
+
+
+@pytest.mark.parametrize(
+    "args,expected_query",
+    [
+        (
+            ["dbt", "execute", "compile", "--name=pipeline_name"],
+            "EXECUTE DBT pipeline_name compile",
+        ),
+        (
+            [
+                "dbt",
+                "execute",
+                "compile",
+                "--name=pipeline_name",
+                "-f",
+                "--select @source:snowplow,tag:nightly models/export",
+            ],
+            "EXECUTE DBT pipeline_name compile -f --select @source:snowplow,tag:nightly models/export",
+        ),
+        (
+            ["dbt", "execute", "compile", "--name=pipeline_name", "--vars '{foo:bar}'"],
+            "EXECUTE DBT pipeline_name compile --vars '{foo:bar}'",
+        ),
+        (
+            [
+                "dbt",
+                "execute",
+                "compile",
+                "--name=pipeline_name",
+                "--format=JSON",
+                "--debug",
+            ],
+            "EXECUTE DBT pipeline_name compile",
+        ),
+    ],
+)
+def test_dbt_execute(mock_connect, runner, args, expected_query):
+
+    result = runner.invoke(args)
+
+    assert result.exit_code == 0, result.output
+    assert mock_connect.mocked_ctx.get_query() == expected_query