Compare commits

...

4 Commits

Author SHA1 Message Date
Gerda Shank
0281b237d6 Fix test 2023-01-03 13:54:00 -05:00
Gerda Shank
b837bf6ef1 Move flag object code to flags.py 2023-01-03 13:08:37 -05:00
Gerda Shank
1c1b119588 Turn context flags into an object instead of return package 2023-01-03 12:58:22 -05:00
Gerda Shank
a911f650b8 Fix overly broad import of os 2023-01-03 12:04:59 -05:00
3 changed files with 33 additions and 14 deletions

View File

@@ -634,9 +634,8 @@ class BaseContext(metaclass=ContextMeta):
{% endif %}
This supports all flags defined in flags submodule (core/dbt/flags.py)
TODO: Replace with object that provides read-only access to flag values
"""
return flags
return flags.get_flag_obj()
@contextmember
@staticmethod

View File

@@ -1,7 +1,9 @@
import os
# Do not import the os package because we expose this package in jinja
from os import name as os_name, path as os_path, getcwd as os_getcwd, getenv as os_getenv
import multiprocessing
from argparse import Namespace
if os.name != "nt":
if os_name != "nt":
# https://bugs.python.org/issue41567
import multiprocessing.popen_spawn_posix # type: ignore
from pathlib import Path
@@ -10,14 +12,14 @@ from typing import Optional
# PROFILES_DIR must be set before the other flags
# It also gets set in main.py and in set_from_args because the rpc server
# doesn't go through exactly the same main arg processing.
GLOBAL_PROFILES_DIR = os.path.join(os.path.expanduser("~"), ".dbt")
LOCAL_PROFILES_DIR = os.getcwd()
GLOBAL_PROFILES_DIR = os_path.join(os_path.expanduser("~"), ".dbt")
LOCAL_PROFILES_DIR = os_getcwd()
# Use the current working directory if there is a profiles.yml file present there
if os.path.exists(Path(LOCAL_PROFILES_DIR) / Path("profiles.yml")):
if os_path.exists(Path(LOCAL_PROFILES_DIR) / Path("profiles.yml")):
DEFAULT_PROFILES_DIR = LOCAL_PROFILES_DIR
else:
DEFAULT_PROFILES_DIR = GLOBAL_PROFILES_DIR
PROFILES_DIR = os.path.expanduser(os.getenv("DBT_PROFILES_DIR", DEFAULT_PROFILES_DIR))
PROFILES_DIR = os_path.expanduser(os_getenv("DBT_PROFILES_DIR", DEFAULT_PROFILES_DIR))
STRICT_MODE = False # Only here for backwards compatibility
FULL_REFRESH = False # subcommand
@@ -88,7 +90,7 @@ def env_set_truthy(key: str) -> Optional[str]:
"""Return the value if it was set to a "truthy" string value or None
otherwise.
"""
value = os.getenv(key)
value = os_getenv(key)
if not value or value.lower() in ("0", "false", "f"):
return None
return value
@@ -101,7 +103,7 @@ def env_set_bool(env_value):
def env_set_path(key: str) -> Optional[Path]:
value = os.getenv(key)
value = os_getenv(key)
if value is None:
return value
else:
@@ -181,7 +183,7 @@ def get_flag_value(flag, args, user_config):
if flag == "PRINTER_WIDTH": # must be ints
flag_value = int(flag_value)
if flag == "PROFILES_DIR":
flag_value = os.path.abspath(flag_value)
flag_value = os_path.abspath(flag_value)
return flag_value
@@ -205,7 +207,7 @@ def _load_flag_value(flag, args, user_config):
def _get_flag_value_from_env(flag):
# Environment variables use pattern 'DBT_{flag name}'
env_flag = _get_env_flag(flag)
env_value = os.getenv(env_flag)
env_value = os_getenv(env_flag)
if env_value is None or env_value == "":
return None
@@ -241,4 +243,21 @@ def get_flag_dict():
"log_cache_events": LOG_CACHE_EVENTS,
"quiet": QUIET,
"no_print": NO_PRINT,
"cache_selected_only": CACHE_SELECTED_ONLY,
"target_path": TARGET_PATH,
"log_path": LOG_PATH,
}
# This is used by core/dbt/context/base.py to return a flag object
# in Jinja.
def get_flag_obj():
new_flags = Namespace()
for k, v in get_flag_dict().items():
setattr(new_flags, k.upper(), v)
# The following 3 are CLI arguments only so they're not full-fledged flags,
# but we put in flags for users.
setattr(new_flags, "FULL_REFRESH", FULL_REFRESH)
setattr(new_flags, "STORE_FAILURES", STORE_FAILURES)
setattr(new_flags, "WHICH", WHICH)
return new_flags

View File

@@ -112,8 +112,9 @@ class TestContextBuiltins:
expected = "invocation_result: {'debug': True, 'log_format': 'json', 'write_json': True, 'use_colors': True, 'printer_width': 80, 'version_check': True, 'partial_parse': True, 'static_parser': True, 'profiles_dir': "
assert expected in str(result)
expected = "'send_anonymous_usage_stats': False, 'quiet': False, 'no_print': False, 'macro': 'validate_invocation', 'args': '{my_variable: test_variable}', 'which': 'run-operation', 'rpc_method': 'run-operation', 'indirect_selection': 'eager'}"
assert expected in str(result)
expected = ("'send_anonymous_usage_stats': False", "'quiet': False", "'no_print': False", "'cache_selected_only': False", "'macro': 'validate_invocation'", "'args': '{my_variable: test_variable}'", "'which': 'run-operation'", "'rpc_method': 'run-operation'", "'indirect_selection': 'eager'")
for element in expected:
assert element in str(result)
def test_builtin_dbt_metadata_envs_function(self, project, monkeypatch):
envs = {