__init__.py

from __future__ import annotations

import os
import warnings
from collections import defaultdict
from json import JSONDecoder
from typing import Any
from typing import Iterator
from typing import TYPE_CHECKING


if TYPE_CHECKING:  # pragma: no cover
    from dynaconf.utils.boxing import DynaBox
    from dynaconf.base import LazySettings, Settings


BANNER = """
██████╗ ██╗   ██╗███╗   ██╗ █████╗  ██████╗ ██████╗ ███╗   ██╗███████╗
██╔══██╗╚██╗ ██╔╝████╗  ██║██╔══██╗██╔════╝██╔═══██╗████╗  ██║██╔════╝
██║  ██║ ╚████╔╝ ██╔██╗ ██║███████║██║     ██║   ██║██╔██╗ ██║█████╗
██║  ██║  ╚██╔╝  ██║╚██╗██║██╔══██║██║     ██║   ██║██║╚██╗██║██╔══╝
██████╔╝   ██║   ██║ ╚████║██║  ██║╚██████╗╚██████╔╝██║ ╚████║██║
╚═════╝    ╚═╝   ╚═╝  ╚═══╝╚═╝  ╚═╝ ╚═════╝ ╚═════╝ ╚═╝  ╚═══╝╚═╝
"""

if os.name == "nt":  # pragma: no cover
    # windows can't handle the above charmap
    BANNER = "DYNACONF"


def object_merge(
    old: Any, new: Any, unique: bool = False, full_path: list[str] = None
) -> Any:
    """
    Recursively merge two data structures, new is mutated in-place.

    :param old: The existing data.
    :param new: The new data to get old values merged in to.
    :param unique: When set to True existing list items are not set.
    :param full_path: Indicates the elements of a tree.
    """
    if full_path is None:
        full_path = []
    if old == new or old is None or new is None:
        # Nothing to merge
        return new

    if isinstance(old, list) and isinstance(new, list):
        for item in old[::-1]:
            if unique and item in new:
                continue
            new.insert(0, item)

    if isinstance(old, dict) and isinstance(new, dict):
        existing_value = recursive_get(old, full_path)  # doesnt handle None
        # Need to make every `None` on `_store` to be an wrapped `LazyNone`

        # data coming from source, in `new` can be mix case: KEY4|key4|Key4
        # data existing on `old` object has the correct case: key4|KEY4|Key4
        # So we need to ensure that new keys matches the existing keys
        for new_key in list(new.keys()):
            correct_case_key = find_the_correct_casing(new_key, old)
            if correct_case_key:
                new[correct_case_key] = new.pop(new_key)

        for old_key, value in old.items():

            # This is for when the dict exists internally
            # but the new value on the end of full path is the same
            if (
                existing_value is not None
                and old_key.lower() == full_path[-1].lower()
                and existing_value is value
            ):
                # Here Be The Dragons
                # This comparison needs to be smarter
                continue

            if old_key not in new:
                new[old_key] = value
            else:
                object_merge(
                    value,
                    new[old_key],
                    full_path=full_path[1:] if full_path else None,
                )

        handle_metavalues(old, new)

    return new


def recursive_get(
    obj: DynaBox | dict[str, int] | dict[str, str | int],
    names: list[str] | None,
) -> Any:
    """Given a dot accessible object and a list of names `foo.bar.zaz`
    gets recursivelly all names one by one obj.foo.bar.zaz.
    """
    if not names:
        return
    head, *tail = names
    result = getattr(obj, head, None)
    if not tail:
        return result
    return recursive_get(result, tail)


def handle_metavalues(
    old: DynaBox | dict[str, int] | dict[str, str | int], new: Any
) -> None:
    """Cleanup of MetaValues on new dict"""

    for key in list(new.keys()):

        # MetaValue instances
        if getattr(new[key], "_dynaconf_reset", False):  # pragma: no cover
            # a Reset on `new` triggers reasign of existing data
            new[key] = new[key].unwrap()
        elif getattr(new[key], "_dynaconf_del", False):
            # a Del on `new` triggers deletion of existing data
            new.pop(key, None)
            old.pop(key, None)
        elif getattr(new[key], "_dynaconf_merge", False):
            # a Merge on `new` triggers merge with existing data
            new[key] = object_merge(
                old.get(key), new[key].unwrap(), unique=new[key].unique
            )

        # Data structures containing merge tokens
        if isinstance(new.get(key), (list, tuple)):
            has_merge = "dynaconf_merge" in new[key]
            has_merge_unique = "dynaconf_merge_unique" in new[key]
            if has_merge or has_merge_unique:
                value = list(new[key])
                unique = False

                try:
                    value.remove("dynaconf_merge")
                except ValueError:
                    value.remove("dynaconf_merge_unique")
                    unique = True

                for item in old.get(key)[::-1]:
                    if unique and item in value:
                        continue
                    value.insert(0, item)

                new[key] = value

        elif isinstance(new.get(key), dict):
            local_merge = new[key].pop(
                "dynaconf_merge", new[key].pop("dynaconf_merge_unique", None)
            )
            if local_merge not in (True, False, None) and not new[key]:
                # In case `dynaconf_merge:` holds value not boolean - ref #241
                new[key] = local_merge

            if local_merge:
                new[key] = object_merge(old.get(key), new[key])


class DynaconfDict(dict):
    """A dict representing en empty Dynaconf object
    useful to run loaders in to a dict for testing"""

    def __init__(self, *args, **kwargs):
        self._fresh = False
        self._loaded_envs = []
        self._loaded_hooks = defaultdict(dict)
        self._loaded_py_modules = []
        self._loaded_files = []
        self._deleted = set()
        self._store = {}
        self._env_cache = {}
        self._loaded_by_loaders = {}
        self._loaders = []
        self._defaults = {}
        self.environ = os.environ
        self.SETTINGS_MODULE = None
        self.filter_strategy = kwargs.get("filter_strategy", None)
        self._not_installed_warnings = []
        self._validate_only = kwargs.pop("validate_only", None)
        self._validate_exclude = kwargs.pop("validate_exclude", None)
        super().__init__(*args, **kwargs)

    def set(self, key: str, value: str, *args, **kwargs) -> None:
        self[key] = value

    @staticmethod
    def get_environ(key, default=None):  # pragma: no cover
        return os.environ.get(key, default)

    def exists(self, key: str, **kwargs) -> bool:
        return self.get(key, missing) is not missing


RENAMED_VARS = {
    # old: new
    "DYNACONF_NAMESPACE": "ENV_FOR_DYNACONF",
    "NAMESPACE_FOR_DYNACONF": "ENV_FOR_DYNACONF",
    "DYNACONF_SETTINGS_MODULE": "SETTINGS_FILE_FOR_DYNACONF",
    "DYNACONF_SETTINGS": "SETTINGS_FILE_FOR_DYNACONF",
    "SETTINGS_MODULE": "SETTINGS_FILE_FOR_DYNACONF",
    "SETTINGS_MODULE_FOR_DYNACONF": "SETTINGS_FILE_FOR_DYNACONF",
    "PROJECT_ROOT": "ROOT_PATH_FOR_DYNACONF",
    "PROJECT_ROOT_FOR_DYNACONF": "ROOT_PATH_FOR_DYNACONF",
    "DYNACONF_SILENT_ERRORS": "SILENT_ERRORS_FOR_DYNACONF",
    "DYNACONF_ALWAYS_FRESH_VARS": "FRESH_VARS_FOR_DYNACONF",
    "BASE_NAMESPACE_FOR_DYNACONF": "DEFAULT_ENV_FOR_DYNACONF",
    "GLOBAL_ENV_FOR_DYNACONF": "ENVVAR_PREFIX_FOR_DYNACONF",
}


def compat_kwargs(kwargs: dict[str, Any]) -> None:
    """To keep backwards compat change the kwargs to new names"""
    warn_deprecations(kwargs)
    for old, new in RENAMED_VARS.items():
        if old in kwargs:
            kwargs[new] = kwargs[old]
            # update cross references
            for c_old, c_new in RENAMED_VARS.items():
                if c_new == new:
                    kwargs[c_old] = kwargs[new]


class Missing:
    """
    Sentinel value object/singleton used to differentiate between ambiguous
    situations where `None` is a valid value.
    """

    def __bool__(self) -> bool:
        """Respond to boolean duck-typing."""
        return False

    def __eq__(self, other: DynaBox | Missing) -> bool:
        """Equality check for a singleton."""

        return isinstance(other, self.__class__)

    # Ensure compatibility with Python 2.x
    __nonzero__ = __bool__

    def __repr__(self) -> str:
        """
        Unambiguously identify this string-based representation of Missing,
        used as a singleton.
        """
        return "<dynaconf.missing>"


missing = Missing()


def deduplicate(list_object: list[str]) -> list[str]:
    """Rebuild `list_object` removing duplicated and keeping order"""
    new = []
    for item in list_object:
        if item not in new:
            new.append(item)
    return new


def warn_deprecations(data: Any) -> None:
    for old, new in RENAMED_VARS.items():
        if old in data:
            warnings.warn(
                f"You are using {old} which is a deprecated settings "
                f"replace it with {new}",
                DeprecationWarning,
            )


def trimmed_split(
    s: str, seps: str | tuple[str, str] = (";", ",")
) -> list[str]:
    """Given a string s, split is by one of one of the seps."""
    for sep in seps:
        if sep not in s:
            continue
        data = [item.strip() for item in s.strip().split(sep)]
        return data
    return [s]  # raw un-splitted


def ensure_a_list(data: Any) -> list[int] | list[str]:
    """Ensure data is a list or wrap it in a list"""
    if not data:
        return []
    if isinstance(data, (list, tuple, set)):
        return list(data)
    if isinstance(data, str):
        data = trimmed_split(data)  # settings.toml,other.yaml
        return data
    return [data]


def build_env_list(obj: Settings | LazySettings, env: str | None) -> list[str]:
    """Build env list for loaders to iterate.

    Arguments:
        obj {LazySettings} -- A Dynaconf settings instance
        env {str} -- The current env to be loaded

    Returns:
        [str] -- A list of string names of the envs to load.
    """
    # add the [default] env
    env_list = [(obj.get("DEFAULT_ENV_FOR_DYNACONF") or "default").lower()]

    # compatibility with older versions that still uses [dynaconf] as
    # [default] env
    global_env = (obj.get("ENVVAR_PREFIX_FOR_DYNACONF") or "dynaconf").lower()
    if global_env not in env_list:
        env_list.append(global_env)

    # add the current env
    current_env = obj.current_env
    if current_env and current_env.lower() not in env_list:
        env_list.append(current_env.lower())

    # add a manually set env
    if env and env.lower() not in env_list:
        env_list.append(env.lower())

    # add the [global] env
    env_list.append("global")

    return env_list


def upperfy(key: str) -> str:
    """Receive a string key and returns its upper version.

    Example:

      input: foo
      output: FOO

      input: foo_bar
      output: FOO_BAR

      input: foo__bar__ZAZ
      output: FOO__bar__ZAZ

    Arguments:
        key {str} -- A string key that may contain dunders `__`

    Returns:
        The key as upper case but keeping the nested elements.
    """
    key = str(key)
    if "__" in key:
        parts = key.split("__")
        return "__".join([parts[0].upper()] + parts[1:])
    return key.upper()


def multi_replace(text: str, patterns: dict[str, str]) -> str:
    """Replaces multiple pairs in a string

    Arguments:
        text {str} -- A "string text"
        patterns {dict} -- A dict of {"old text": "new text"}

    Returns:
        text -- str
    """
    for old, new in patterns.items():
        text = text.replace(old, new)
    return text


def extract_json_objects(
    text: str, decoder: JSONDecoder = JSONDecoder()
) -> Iterator[dict[str, int | dict[Any, Any]]]:
    """Find JSON objects in text, and yield the decoded JSON data

    Does not attempt to look for JSON arrays, text, or other JSON types outside
    of a parent JSON object.

    """
    pos = 0
    while True:
        match = text.find("{", pos)
        if match == -1:
            break
        try:
            result, index = decoder.raw_decode(text[match:])
            yield result
            pos = match + index
        except ValueError:
            pos = match + 1


def recursively_evaluate_lazy_format(
    value: Any, settings: Settings | LazySettings
) -> Any:
    """Given a value as a data structure, traverse all its members
    to find Lazy values and evaluate it.

    For example: Evaluate values inside lists and dicts
    """

    if getattr(value, "_dynaconf_lazy_format", None):
        value = value(settings)

    if isinstance(value, list):
        # Keep the original type, can be a BoxList
        value = value.__class__(
            [
                recursively_evaluate_lazy_format(item, settings)
                for item in value
            ]
        )

    return value


def isnamedtupleinstance(value):
    """Check if value is a namedtuple instance

    stackoverflow.com/questions/2166818/
    how-to-check-if-an-object-is-an-instance-of-a-namedtuple
    """

    t = type(value)
    b = t.__bases__
    if len(b) != 1 or b[0] != tuple:
        return False
    f = getattr(t, "_fields", None)
    if not isinstance(f, tuple):
        return False
    return all(type(n) == str for n in f)


def find_the_correct_casing(key: str, data: dict[str, Any]) -> str | None:
    """Given a key, find the proper casing in data

    Arguments:
        key {str} -- A key to be searched in data
        data {dict} -- A dict to be searched

    Returns:
        str -- The proper casing of the key in data
    """
    if key in data:
        return key
    for k in data.keys():
        if k.lower() == key.lower():
            return k
    return None