Handle dbt kwargs as args (#18952)

Author: kevingrismore

src/integrations/prefect-dbt/prefect_dbt/core/runner.py

@@ -38,7 +38,7 @@
 from prefect.tasks import MaterializingTask, Task, TaskOptions
 from prefect_dbt.core._tracker import NodeTaskTracker
 from prefect_dbt.core.settings import PrefectDbtSettings
-from prefect_dbt.utilities import format_resource_id
+from prefect_dbt.utilities import format_resource_id, kwargs_to_args
 
 FAILURE_STATUSES = [
     RunStatus.Error,
@@ -660,8 +660,7 @@ def invoke(self, args: list[str], **kwargs: Any):
             invoke_kwargs["profiles_dir"] = profiles_dir
 
             res = dbtRunner(callbacks=callbacks).invoke(  # type: ignore[reportUnknownMemberType]
-                args_copy,
-                **invoke_kwargs,
+                kwargs_to_args(invoke_kwargs, args_copy)
             )
 
         if not res.success and res.exception:

src/integrations/prefect-dbt/prefect_dbt/utilities.py

@@ -4,7 +4,7 @@
 
 import os
 from pathlib import Path
-from typing import Any
+from typing import Any, Optional
 
 import slugify
 
@@ -71,3 +71,64 @@ def format_resource_id(adapter_type: str, relation_name: str) -> str:
         asset_key = asset_key[:MAX_ASSET_KEY_LENGTH]
 
     return asset_key
+
+
+def kwargs_to_args(kwargs: dict, args: Optional[list[str]] = None) -> list[str]:
+    """
+    Convert a dictionary of kwargs to a list of args in the dbt CLI format.
+    If args are provided, they take priority over kwargs when conflicts exist.
+
+    Args:
+        kwargs: A dictionary of kwargs.
+        args: Optional list of existing args that take priority over kwargs.
+
+    Returns:
+        A list of args.
+    """
+    if args is None:
+        args = []
+
+    kwargs_args = []
+    for key, value in list(kwargs.items()):
+        if value is None:
+            continue
+
+        flag = "--" + key.replace("_", "-")
+
+        if isinstance(value, bool):
+            kwargs_args.append(flag if value else f"--no-{key.replace('_', '-')}")
+
+        elif isinstance(value, (list, tuple)):
+            kwargs_args.append(flag)
+            kwargs_args.extend(str(v) for v in value)
+        else:
+            kwargs_args.extend([flag, str(value)])
+
+        kwargs.pop(key)
+
+    existing_flags = _parse_args_to_flag_groups(args)
+    kwargs_flags = _parse_args_to_flag_groups(kwargs_args)
+
+    result_args = []
+    for flag, values in existing_flags.items():
+        result_args.extend([flag] + values)
+
+    for flag, values in kwargs_flags.items():
+        if flag not in existing_flags:
+            result_args.extend([flag] + values)
+
+    return result_args
+
+
+def _parse_args_to_flag_groups(args_list: list[str]) -> dict[str, list[str]]:
+    groups = {}
+    current_flag = None
+
+    for arg in args_list:
+        if arg.startswith("--"):
+            current_flag = arg
+            groups[current_flag] = []
+        elif current_flag:
+            groups[current_flag].append(arg)
+
+    return groups

src/integrations/prefect-dbt/tests/core/test_runner.py

@@ -518,7 +518,7 @@ def test_flow():
 
         # Verify log_level was set to "none"
         call_args = mock_dbt_runner_class.return_value.invoke.call_args
-        assert call_args[1]["log_level"] == "none"
+        assert "--log-level", "none" in call_args[0]
 
     def test_invoke_uses_original_log_level_outside_context(
         self, mock_dbt_runner_class, mock_settings_context_manager
@@ -535,7 +535,7 @@ def test_invoke_uses_original_log_level_outside_context(
 
         # Verify log_level was set to the original value
         call_args = mock_dbt_runner_class.return_value.invoke.call_args
-        assert call_args[1]["log_level"] == str(runner.log_level.value)
+        assert "--log-level", str(runner.log_level.value) in call_args[0]
 
     def test_invoke_handles_dbt_exceptions(
         self, mock_dbt_runner_class, mock_settings_context_manager
@@ -609,7 +609,8 @@ def test_invoke_handles_kwargs_and_cli_flags_together(
         assert result.success is True
         # Verify the CLI flags take precedence (processed after kwargs)
         call_args = mock_dbt_runner_class.return_value.invoke.call_args
-        assert call_args[1]["target_path"] == "/cli/path"
+        assert "--target-path", "/cli/path" in call_args[0]
+        assert "--target-path", "/kwargs/path" not in call_args[0]
 
     def test_invoke_uses_resolve_profiles_yml_context_manager(
         self, mock_dbt_runner_class, mock_settings_context_manager

src/integrations/prefect-dbt/tests/test_utilities.py

@@ -9,6 +9,7 @@
 from prefect_dbt.utilities import (
     find_profiles_dir,
     format_resource_id,
+    kwargs_to_args,
     replace_with_env_var_call,
 )
 
@@ -206,3 +207,83 @@ def test_format_resource_id_over_max_length(self) -> None:
         expected = MAX_ASSET_KEY_LENGTH
 
         assert result == expected
+
+
+class TestKwargsToArgs:
+    """Test cases for kwargs_to_args function."""
+
+    def test_kwargs_to_args_with_simple_kwargs(self) -> None:
+        """Test with simple kwargs."""
+        kwargs = {"a": 1, "b": 2, "c": 3}
+
+        result = kwargs_to_args(kwargs)
+        expected = ["--a", "1", "--b", "2", "--c", "3"]
+
+        assert result == expected
+
+    def test_kwargs_to_args_with_boolean_kwargs(self) -> None:
+        """Test with boolean kwargs."""
+        kwargs = {"a": True, "b": False, "c": True}
+
+        result = kwargs_to_args(kwargs)
+        expected = ["--a", "--no-b", "--c"]
+
+        assert result == expected
+
+    def test_kwargs_to_args_with_list_kwargs(self) -> None:
+        """Test with list kwargs."""
+        kwargs = {"a": [1, 2, 3], "b": [4, 5, 6]}
+
+        result = kwargs_to_args(kwargs)
+        expected = ["--a", "1", "2", "3", "--b", "4", "5", "6"]
+
+        assert result == expected
+
+    def test_kwargs_to_args_with_existing_args(self) -> None:
+        """Test with existing args that don't conflict with kwargs."""
+        kwargs = {"a": 1, "b": 2}
+        args = ["--c", "3", "--d", "4"]
+
+        result = kwargs_to_args(kwargs, args)
+        expected = ["--c", "3", "--d", "4", "--a", "1", "--b", "2"]
+
+        assert result == expected
+
+    def test_kwargs_to_args_args_priority_over_kwargs(self) -> None:
+        """Test that args take priority over kwargs when conflicts exist."""
+        kwargs = {"a": 1, "b": 2, "c": 3}
+        args = ["--a", "10", "--d", "4"]
+
+        result = kwargs_to_args(kwargs, args)
+        expected = ["--a", "10", "--d", "4", "--b", "2", "--c", "3"]
+
+        assert result == expected
+
+    def test_kwargs_to_args_boolean_conflict(self) -> None:
+        """Test boolean flag conflicts between args and kwargs."""
+        kwargs = {"a": True, "b": False}
+        args = ["--a", "--no-b"]
+
+        result = kwargs_to_args(kwargs, args)
+        expected = ["--a", "--no-b"]
+
+        assert result == expected
+
+    def test_kwargs_to_args_list_conflict(self) -> None:
+        """Test list flag conflicts between args and kwargs."""
+        kwargs = {"a": [1, 2, 3], "b": [4, 5]}
+        args = ["--a", "10", "20"]
+
+        result = kwargs_to_args(kwargs, args)
+        expected = ["--a", "10", "20", "--b", "4", "5"]
+
+        assert result == expected
+
+    def test_kwargs_to_args_no_args_provided(self) -> None:
+        """Test behavior when no args are provided (backward compatibility)."""
+        kwargs = {"a": 1, "b": 2}
+
+        result = kwargs_to_args(kwargs)
+        expected = ["--a", "1", "--b", "2"]
+
+        assert result == expected