You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
bazarr/libs/dynaconf/loaders/__init__.py

383 lines
12 KiB

from __future__ import annotations
import importlib
import os
from typing import Callable
from typing import TYPE_CHECKING
from dynaconf import constants as ct
from dynaconf import default_settings
from dynaconf.loaders import ini_loader
from dynaconf.loaders import json_loader
from dynaconf.loaders import py_loader
from dynaconf.loaders import toml_loader
from dynaconf.loaders import yaml_loader
from dynaconf.utils import deduplicate
from dynaconf.utils import ensure_a_list
from dynaconf.utils.boxing import DynaBox
from dynaconf.utils.files import get_local_filename
from dynaconf.utils.files import glob
from dynaconf.utils.files import has_magic
from dynaconf.utils.parse_conf import false_values
if TYPE_CHECKING:
from dynaconf.base import Settings
def default_loader(obj, defaults=None):
"""Initial loader for the initialization process.
Steps:
- Load default settings (from static module) + kwargs overrides (togheter)
- Load envvar overrides
"""
# LOAD DEFAULT STATIC + KWARGS OVERRIDES
defaults = defaults or {}
default_settings_values = {
key: value
for key, value in default_settings.__dict__.items() # noqa
if key.isupper()
}
all_keys = deduplicate(
list(defaults.keys()) + list(default_settings_values.keys())
)
for key in all_keys:
if not obj.exists(key):
value = defaults.get(key, default_settings_values.get(key))
obj.set(
key,
value,
loader_identifier="default_settings",
validate=False,
)
# LOAD ENVVAR OVERRIDES
# start dotenv to get default env vars from there
# check overrides in env vars
if obj.get("load_dotenv") is True:
default_settings.start_dotenv(obj)
# Deal with cases where a custom ENV_SWITCHER_IS_PROVIDED
# Example: Flask and Django Extensions
env_switcher = defaults.get(
"ENV_SWITCHER_FOR_DYNACONF", "ENV_FOR_DYNACONF"
)
for key in all_keys:
if key not in default_settings_values.keys():
continue
env_value = obj.get_environ(
env_switcher if key == "ENV_FOR_DYNACONF" else key,
default="_not_found",
)
if env_value != "_not_found":
obj.set(
key,
env_value,
tomlfy=True,
loader_identifier="envvars_first_load",
)
def execute_instance_hooks(
obj: Settings, hook_type: str, hook_functions: list[Callable]
):
"""Execute hooks provided by Setting instance"""
hook_source = "instance"
for hook_func in hook_functions:
_run_hook_function(obj, hook_type, hook_func, hook_source)
def execute_module_hooks(
hook, obj, env=None, silent=True, key=None, modules=None, files=None
):
"""Execute dynaconf_hooks from module or filepath."""
if hook not in ["post"]:
raise ValueError(f"hook {hook} not supported yet.")
# try to load hooks using python module __name__
modules = modules or obj._loaded_py_modules
for loaded_module in modules:
hook_module_name = ".".join(
loaded_module.split(".")[:-1] + ["dynaconf_hooks"]
)
try:
hook_module = importlib.import_module(hook_module_name)
except (ImportError, TypeError):
# There was no hook on the same path as a python module
continue
else:
_run_hook_module(
hook_type=hook,
hook_module=hook_module,
obj=obj,
key=key,
)
# Try to load from python filename path
files = files or obj._loaded_files
for loaded_file in files:
hook_file = os.path.join(
os.path.dirname(loaded_file), "dynaconf_hooks.py"
)
hook_module = py_loader.import_from_filename(
obj, hook_file, silent=silent
)
if not hook_module:
# There was no hook on the same path as a python file
continue
_run_hook_module(
hook_type=hook,
hook_module=hook_module,
obj=obj,
key=key,
)
# alias
execute_hooks = execute_module_hooks
def _run_hook_module(hook_type, hook_module, obj, key=""):
"""
Run a hook function from hook_module.
Given a @hook_type, a @hook_module and a settings @obj, load the function
and execute it if found.
"""
hook_source = hook_module.__file__
# check if already loaded
if hook_type in obj._loaded_hooks.get(hook_source, {}):
return
# check errors
if hook_module and getattr(hook_module, "_error", False):
if not isinstance(hook_module._error, FileNotFoundError):
raise hook_module._error
# execute hook
hook_func = getattr(hook_module, hook_type, None)
if hook_func:
_run_hook_function(obj, hook_type, hook_func, hook_source, key)
def _run_hook_function(
obj: Settings,
hook_type: str,
hook_func: Callable,
hook_source: str = "default",
key: str = "",
):
"""
Run a hook function:
It execute @hook_func, update the results into settings @obj and
add it to _loaded_hook registry ([@hook_source][@hook_type])
"""
# optional settings argument
try:
hook_dict = hook_func(obj.dynaconf.clone())
except TypeError:
hook_dict = hook_func()
# update obj settings
if hook_dict:
identifier = f"{hook_func.__name__}@{hook_source}"
merge = hook_dict.pop(
"dynaconf_merge", hook_dict.pop("DYNACONF_MERGE", False)
)
if key and key in hook_dict:
obj.set(
key,
hook_dict[key],
tomlfy=False,
merge=merge,
loader_identifier=identifier,
)
elif not key:
obj.update(
hook_dict,
tomlfy=False,
merge=merge,
loader_identifier=identifier,
)
# add to registry
obj._loaded_hooks[hook_source][hook_type] = hook_dict
def settings_loader(
obj,
settings_module=None,
env=None,
silent=True,
key=None,
filename=None,
validate=False,
):
"""Loads from defined settings module
:param obj: A dynaconf instance
:param settings_module: A path or a list of paths e.g settings.toml
:param env: Env to look for data defaults: development
:param silent: Boolean to raise loading errors
:param key: Load a single key if provided
:param filename: optional filename to override the settings_module
"""
if filename is None:
settings_module = settings_module or obj.settings_module
if not settings_module: # pragma: no cover
return
files = ensure_a_list(settings_module)
else:
files = ensure_a_list(filename)
files.extend(ensure_a_list(obj.get("SECRETS_FOR_DYNACONF", None)))
found_files = []
modules_names = []
for item in files:
item = str(item) # Ensure str in case of LocalPath/Path is passed.
p_root = obj._root_path or (
os.path.dirname(found_files[0]) if found_files else None
)
if has_magic(item):
# handle possible globs inside files list
# like ["path/*.yaml", "path/ABC?.yaml"]
globedfiles = glob(item, root_dir=p_root)
for globedfile in globedfiles:
# use object.find_file logic to handle skip files
found = obj.find_file(globedfile, project_root=p_root)
if found:
found_files.append(found)
elif item.endswith(ct.ALL_EXTENSIONS + (".py",)):
found = obj.find_file(item, project_root=p_root)
if found:
found_files.append(found)
else:
# a bare python module name w/o extension
modules_names.append(item)
enabled_core_loaders = [
item.upper() for item in obj.get("CORE_LOADERS_FOR_DYNACONF") or []
]
# add `.local.` to found_files list to search for local files.
found_files.extend(
[
get_local_filename(item)
for item in found_files
if ".local." not in str(item)
]
)
for mod_file in modules_names + found_files:
# can be set to multiple files settings.py,settings.yaml,...
# Cascade all loaders
loaders = [
{"ext": ct.YAML_EXTENSIONS, "name": "YAML", "loader": yaml_loader},
{"ext": ct.TOML_EXTENSIONS, "name": "TOML", "loader": toml_loader},
{"ext": ct.INI_EXTENSIONS, "name": "INI", "loader": ini_loader},
{"ext": ct.JSON_EXTENSIONS, "name": "JSON", "loader": json_loader},
]
for loader in loaders:
if loader["name"] not in enabled_core_loaders:
continue
if mod_file.endswith(loader["ext"]):
loader["loader"].load(
obj,
filename=mod_file,
env=env,
silent=silent,
key=key,
validate=validate,
)
continue
if mod_file.endswith(ct.ALL_EXTENSIONS):
continue
if "PY" not in enabled_core_loaders:
# pyloader is disabled
continue
# must be Python file or module
# load from default defined module settings.py or .secrets.py if exists
py_loader.load(obj, mod_file, key=key, validate=validate)
# load from the current env e.g: development_settings.py
env = env or obj.current_env
if mod_file.endswith(".py"):
if ".secrets.py" == mod_file:
tmpl = ".{0}_{1}{2}"
mod_file = "secrets.py"
else:
tmpl = "{0}_{1}{2}"
dirname = os.path.dirname(mod_file)
filename, extension = os.path.splitext(os.path.basename(mod_file))
new_filename = tmpl.format(env.lower(), filename, extension)
env_mod_file = os.path.join(dirname, new_filename)
global_filename = tmpl.format("global", filename, extension)
global_mod_file = os.path.join(dirname, global_filename)
else:
env_mod_file = f"{env.lower()}_{mod_file}"
global_mod_file = f"global_{mod_file}"
py_loader.load(
obj,
env_mod_file,
identifier=f"py_{env.upper()}",
silent=True,
key=key,
validate=validate,
)
# load from global_settings.py
py_loader.load(
obj,
global_mod_file,
identifier="py_global",
silent=True,
key=key,
validate=validate,
)
def enable_external_loaders(obj):
"""Enable external service loaders like `VAULT_` and `REDIS_`
looks forenv variables like `REDIS_ENABLED_FOR_DYNACONF`
"""
for name, loader in ct.EXTERNAL_LOADERS.items():
enabled = getattr(obj, f"{name.upper()}_ENABLED_FOR_DYNACONF", False)
if (
enabled
and enabled not in false_values
and loader not in obj.LOADERS_FOR_DYNACONF
): # noqa
obj.LOADERS_FOR_DYNACONF.insert(0, loader)
def write(filename, data, env=None, merge=False):
"""Writes `data` to `filename` infers format by file extension."""
loader_name = f"{filename.rpartition('.')[-1]}_loader"
loader = globals().get(loader_name)
if not loader:
raise OSError(f"{loader_name} cannot be found.")
data = DynaBox(data, box_settings={}).to_dict()
if loader is not py_loader and env and env not in data:
data = {env: data}
loader.write(filename, data, merge=merge)