Compare commits

...

5 Commits

Author SHA1 Message Date
Doug Beatty
1e53c7d90b Remove unused variable e 2025-06-27 13:24:45 -06:00
Doug Beatty
2b76da0e1a Accommodate kwargs with False boolean flags 2025-06-27 10:48:24 -06:00
Doug Beatty
3c4341ce3b Fix variable name 2025-06-27 10:46:06 -06:00
Doug Beatty
751deb69ca Changelog entry 2025-06-27 09:33:42 -06:00
Doug Beatty
4dc7570624 Set flags.INVOCATION_COMMAND for programmatic invocations on a best-effort basis 2025-06-27 08:48:21 -06:00
3 changed files with 54 additions and 2 deletions

View File

@@ -0,0 +1,6 @@
kind: Fixes
body: Set `flags.INVOCATION_COMMAND` for programmatic invocations on a best-effort basis
time: 2025-06-27T09:33:25.560217-06:00
custom:
Author: dbeatty10
Issue: "10313"

View File

@@ -261,8 +261,9 @@ class Flags:
else: else:
project_flags = None project_flags = None
# Add entire invocation command to flags # Add entire invocation command to flags on a best-effort basis (either from env var or from sys.argv)
object.__setattr__(self, "INVOCATION_COMMAND", "dbt " + " ".join(sys.argv[1:])) invocation_command = os.getenv("DBT_EQUIVALENT_COMMAND", "dbt " + " ".join(sys.argv[1:]))
object.__setattr__(self, "INVOCATION_COMMAND", invocation_command)
if project_flags: if project_flags:
# Overwrite default assignments with project flags if available. # Overwrite default assignments with project flags if available.

View File

@@ -1,4 +1,6 @@
import functools import functools
import os
import shlex
from copy import copy from copy import copy
from dataclasses import dataclass from dataclasses import dataclass
from typing import Callable, List, Optional, Union from typing import Callable, List, Optional, Union
@@ -37,6 +39,43 @@ class dbtRunnerResult:
] = None ] = None
def get_equivalent_cli_flag(config_name: str, positive_config: bool = True) -> str:
"""Convert a config name to its equivalent CLI flag"""
if positive_config:
return f"--{config_name.replace('_', '-')}"
else:
return f"--no-{config_name.replace('_', '-')}"
def get_equivalent_cli_command(args: List[str], **kwargs) -> str:
"""Convert args and kwargs to a CLI command string that can be used to invoke dbt on a best-effort basis"""
# Convert all values in `args` to strings
cli_args = [str(arg) for arg in args]
# Convert each keyword arg to its CLI flag equivalent
for key, value in kwargs.items():
if type(value) is bool:
# Add just the flag (without the value) for booleans
cli_args.append(get_equivalent_cli_flag(key, value))
elif type(value) in {list, tuple}:
# Add both the flag and its values for lists/tuples
cli_args.append(get_equivalent_cli_flag(key))
cli_args.extend(map(str, value))
else:
# Add both the flag and its value for other types
# This is a best-effort conversion, so we ignore exceptions
try:
cli_args.extend([get_equivalent_cli_flag(key), str(value)])
except Exception:
pass
# Ensure all CLI arguments are quoted as needed
cli_args = [shlex.quote(v) for v in cli_args]
return "dbt " + " ".join(cli_args)
# Programmatic invocation # Programmatic invocation
class dbtRunner: class dbtRunner:
def __init__( def __init__(
@@ -64,6 +103,12 @@ class dbtRunner:
# Hack to set parameter source to custom string # Hack to set parameter source to custom string
dbt_ctx.set_parameter_source(key, "kwargs") # type: ignore dbt_ctx.set_parameter_source(key, "kwargs") # type: ignore
# Get equivalent CLI command from args
equivalent_command = get_equivalent_cli_command(args, **kwargs)
# Store the invocation command in an environment variable for access within `cli/flags.py`
os.environ["DBT_EQUIVALENT_COMMAND"] = equivalent_command
result, success = cli.invoke(dbt_ctx) result, success = cli.invoke(dbt_ctx)
return dbtRunnerResult( return dbtRunnerResult(
result=result, result=result,