flags.py

# Do not import the os package because we expose this package in jinja
from os import getenv as os_getenv
from argparse import Namespace
from multiprocessing import get_context
from typing import Optional


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)
    if not value or value.lower() in ("0", "false", "f"):
        return None
    return value


# for setting up logger for legacy logger
ENABLE_LEGACY_LOGGER = env_set_truthy("DBT_ENABLE_LEGACY_LOGGER")
LOG_FORMAT = None
DEBUG = None
USE_COLORS = None
LOG_CACHE_EVENTS = None
QUIET = None

# This is not a flag, it's a place to store the lock
MP_CONTEXT = get_context()


# this roughly follows the patten of EVENT_MANAGER in dbt/events/functions.py
# During de-globlization, we'll need to handle both similarly
GLOBAL_FLAGS = Namespace()  # type: ignore
FLAGS_SET = False


def set_flags(flags):
    global GLOBAL_FLAGS
    global FLAGS_SET
    FLAGS_SET = True
    GLOBAL_FLAGS = flags


def get_flags():
    global FLAGS_SET
    # this allow use the defualt via get_flags()
    if not FLAGS_SET:
        set_from_args(Namespace(), None)
    return GLOBAL_FLAGS


def set_from_args(args: Namespace, user_config):
    global GLOBAL_FLAGS
    from dbt.cli.main import cli
    from dbt.cli.flags import Flags

    # make a dummy context to get the flags
    ctx = cli.make_context("run", ["run"])
    flags = Flags(ctx, user_config)
    for arg_name, args_param_value in vars(args).items():
        if arg_name in ["warn_error_options"]:
            from dbt.cli.option_types import WarnErrorOptionsType

            args_param_value = WarnErrorOptionsType().convert(args_param_value, None, None)
        object.__setattr__(flags, arg_name.upper(), args_param_value)
        object.__setattr__(flags, arg_name.lower(), args_param_value)
    GLOBAL_FLAGS = flags  # type: ignore
    global FLAGS_SET
    FLAGS_SET = True


def get_flag_dict():
    flag_attr = {
        "use_experimental_parser",
        "static_parser",
        "warn_error",
        "warn_error_options",
        "write_json",
        "partial_parse",
        "use_colors",
        "profiles_dir",
        "debug",
        "log_format",
        "version_check",
        "fail_fast",
        "send_anonymous_usage_stats",
        "anonymous_usage_stats",
        "printer_width",
        "indirect_selection",
        "log_cache_events",
        "quiet",
        "no_print",
        "cache_selected_only",
        "target_path",
        "log_path",
    }
    return {key: getattr(GLOBAL_FLAGS, key.upper(), None) for key in flag_attr}


# This is used by core/dbt/context/base.py to return a flag object
# in Jinja.
def get_flag_obj():
    new_flags = Namespace()
    for key, val in get_flag_dict().items():
        setattr(new_flags, key.upper(), val)
    # 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", getattr(GLOBAL_FLAGS, "FULL_REFRESH", None))
    setattr(new_flags, "STORE_FAILURES", getattr(GLOBAL_FLAGS, "STORE_FAILURES", None))
    setattr(new_flags, "WHICH", getattr(GLOBAL_FLAGS, "WHICH", None))
    return new_flags