Add kwargs support to dbtRunner (#7274)

.changes/unreleased/Under the Hood-20230404-181717.yaml

@@ -0,0 +1,6 @@
+kind: Under the Hood
+body: Add kwargs support to dbtRunner
+time: 2023-04-04T18:17:17.539426-07:00
+custom:
+  Author: aranke
+  Issue: "7070"

core/dbt/cli/flags.py

@@ -112,7 +112,10 @@ def assign_params(ctx, params_assigned_from_default, deprecated_env_vars):
                         # param already set via its deprecated but still respected env var
                         continue
 
-                    if param_name not in EXPECTED_DUPLICATE_PARAMS:
+                    if (
+                        param_name not in EXPECTED_DUPLICATE_PARAMS
+                        and ctx.get_parameter_source(param_name) != "kwargs"
+                    ):
                         raise Exception(
                             f"Duplicate flag names found in click command: {param_name}"
                         )

core/dbt/cli/main.py

@@ -2,27 +2,28 @@
 from typing import Callable, List, Tuple, Optional
 
 import click
+
 from dbt.cli import requires, params as p
-from dbt.config.project import Project
 from dbt.config.profile import Profile
+from dbt.config.project import Project
 from dbt.contracts.graph.manifest import Manifest
 from dbt.events.base_types import EventMsg
+from dbt.task.build import BuildTask
 from dbt.task.clean import CleanTask
 from dbt.task.compile import CompileTask
-from dbt.task.deps import DepsTask
 from dbt.task.debug import DebugTask
+from dbt.task.deps import DepsTask
+from dbt.task.freshness import FreshnessTask
+from dbt.task.generate import GenerateTask
+from dbt.task.init import InitTask
+from dbt.task.list import ListTask
 from dbt.task.run import RunTask
+from dbt.task.run_operation import RunOperationTask
+from dbt.task.seed import SeedTask
 from dbt.task.serve import ServeTask
 from dbt.task.show import ShowTask
-from dbt.task.test import TestTask
 from dbt.task.snapshot import SnapshotTask
-from dbt.task.seed import SeedTask
-from dbt.task.list import ListTask
-from dbt.task.freshness import FreshnessTask
-from dbt.task.run_operation import RunOperationTask
-from dbt.task.build import BuildTask
-from dbt.task.generate import GenerateTask
-from dbt.task.init import InitTask
+from dbt.task.test import TestTask
 
 
 class dbtUsageException(Exception):
@@ -47,7 +48,7 @@ def __init__(
         self.manifest = manifest
         self.callbacks = callbacks
 
-    def invoke(self, args: List[str]) -> Tuple[Optional[List], bool]:
+    def invoke(self, args: List[str], **kwargs) -> Tuple[Optional[List], bool]:
         try:
             dbt_ctx = cli.make_context(cli.name, args)
             dbt_ctx.obj = {
@@ -56,6 +57,12 @@ def invoke(self, args: List[str]) -> Tuple[Optional[List], bool]:
                 "manifest": self.manifest,
                 "callbacks": self.callbacks,
             }
+
+            for key, value in kwargs.items():
+                dbt_ctx.params[key] = value
+                # Hack to set parameter source to custom string
+                dbt_ctx.set_parameter_source(key, "kwargs")  # type: ignore
+
             return cli.invoke(dbt_ctx)
         except requires.HandledExit as e:
             return (e.result, e.success)

tests/functional/dbt_runner/test_dbt_runner.py

@@ -0,0 +1,69 @@
+from unittest import mock
+
+import pytest
+
+from dbt.cli.main import dbtRunner, dbtUsageException
+from dbt.exceptions import DbtProjectError
+
+
+class TestDbtRunner:
+    @pytest.fixture
+    def dbt(self) -> dbtRunner:
+        return dbtRunner()
+
+    def test_group_invalid_option(self, dbt: dbtRunner) -> None:
+        with pytest.raises(dbtUsageException):
+            dbt.invoke(["--invalid-option"])
+
+    def test_command_invalid_option(self, dbt: dbtRunner) -> None:
+        with pytest.raises(dbtUsageException):
+            dbt.invoke(["deps", "--invalid-option"])
+
+    def test_command_mutually_exclusive_option(self, dbt: dbtRunner) -> None:
+        with pytest.raises(dbtUsageException):
+            dbt.invoke(["--warn-error", "--warn-error-options", '{"include": "all"}', "deps"])
+
+    def test_invalid_command(self, dbt: dbtRunner) -> None:
+        with pytest.raises(dbtUsageException):
+            dbt.invoke(["invalid-command"])
+
+    def test_invoke_version(self, dbt: dbtRunner) -> None:
+        dbt.invoke(["--version"])
+
+    def test_callbacks(self) -> None:
+        mock_callback = mock.MagicMock()
+        dbt = dbtRunner(callbacks=[mock_callback])
+        # the `debug` command is one of the few commands wherein you don't need
+        # to have a project to run it and it will emit events
+        dbt.invoke(["debug"])
+        mock_callback.assert_called()
+
+    def test_invoke_kwargs(self, project, dbt):
+        results, success = dbt.invoke(
+            ["run"],
+            log_format="json",
+            log_path="some_random_path",
+            version_check=False,
+            profile_name="some_random_profile_name",
+            target_dir="some_random_target_dir",
+        )
+        assert results.args["log_format"] == "json"
+        assert results.args["log_path"] == "some_random_path"
+        assert results.args["version_check"] is False
+        assert results.args["profile_name"] == "some_random_profile_name"
+        assert results.args["target_dir"] == "some_random_target_dir"
+
+    def test_invoke_kwargs_project_dir(self, project, dbt):
+        with pytest.raises(
+            DbtProjectError,
+            match="No dbt_project.yml found at expected path some_random_project_dir",
+        ):
+            dbt.invoke(["run"], project_dir="some_random_project_dir")
+
+    def test_invoke_kwargs_profiles_dir(self, project, dbt):
+        with pytest.raises(DbtProjectError, match="Could not find profile named 'test'"):
+            dbt.invoke(["run"], profiles_dir="some_random_profiles_dir")
+
+    def test_invoke_kwargs_and_flags(self, project, dbt):
+        results, success = dbt.invoke(["--log-format=text", "run"], log_format="json")
+        assert results.args["log_format"] == "json"