diff options
Diffstat (limited to 'libs/dynaconf/loaders/base.py')
| -rw-r--r-- | libs/dynaconf/loaders/base.py | 195 |
1 files changed, 195 insertions, 0 deletions
diff --git a/libs/dynaconf/loaders/base.py b/libs/dynaconf/loaders/base.py new file mode 100644 index 000000000..dec5cb0af --- /dev/null +++ b/libs/dynaconf/loaders/base.py @@ -0,0 +1,195 @@ +from __future__ import annotations + +import io +import warnings + +from dynaconf.utils import build_env_list +from dynaconf.utils import ensure_a_list +from dynaconf.utils import upperfy + + +class BaseLoader: + """Base loader for dynaconf source files. + + :param obj: {[LazySettings]} -- [Dynaconf settings] + :param env: {[string]} -- [the current env to be loaded defaults to + [development]] + :param identifier: {[string]} -- [identifier ini, yaml, json, py, toml] + :param extensions: {[list]} -- [List of extensions with dots ['.a', '.b']] + :param file_reader: {[callable]} -- [reads file return dict] + :param string_reader: {[callable]} -- [reads string return dict] + """ + + def __init__( + self, + obj, + env, + identifier, + extensions, + file_reader, + string_reader, + opener_params=None, + ): + """Instantiates a loader for different sources""" + self.obj = obj + self.env = env or obj.current_env + self.identifier = identifier + self.extensions = extensions + self.file_reader = file_reader + self.string_reader = string_reader + self.opener_params = opener_params or { + "mode": "r", + "encoding": obj.get("ENCODING_FOR_DYNACONF", "utf-8"), + } + + @staticmethod + def warn_not_installed(obj, identifier): # pragma: no cover + if identifier not in obj._not_installed_warnings: + warnings.warn( + f"{identifier} support is not installed in your environment. " + f"`pip install dynaconf[{identifier}]`" + ) + obj._not_installed_warnings.append(identifier) + + def load(self, filename=None, key=None, silent=True): + """ + Reads and loads in to `self.obj` a single key or all keys from source + + :param filename: Optional filename to load + :param key: if provided load a single key + :param silent: if load errors should be silenced + """ + + filename = filename or self.obj.get(self.identifier.upper()) + if not filename: + return + + if not isinstance(filename, (list, tuple)): + split_files = ensure_a_list(filename) + if all([f.endswith(self.extensions) for f in split_files]): # noqa + files = split_files # it is a ['file.ext', ...] + else: # it is a single config as string + files = [filename] + else: # it is already a list/tuple + files = filename + + source_data = self.get_source_data(files) + + if self.obj.get("ENVIRONMENTS_FOR_DYNACONF") is False: + self._envless_load(source_data, silent, key) + else: + self._load_all_envs(source_data, silent, key) + + def get_source_data(self, files): + """Reads each file and returns source data for each file + {"path/to/file.ext": {"key": "value"}} + """ + data = {} + for source_file in files: + if source_file.endswith(self.extensions): + try: + with open(source_file, **self.opener_params) as open_file: + content = self.file_reader(open_file) + self.obj._loaded_files.append(source_file) + if content: + data[source_file] = content + except OSError as e: + if ".local." not in source_file: + warnings.warn( + f"{self.identifier}_loader: {source_file} " + f":{str(e)}" + ) + else: + # for tests it is possible to pass string + content = self.string_reader(source_file) + if content: + data[source_file] = content + return data + + def _envless_load(self, source_data, silent=True, key=None): + """Load all the keys from each file without env separation""" + for file_data in source_data.values(): + self._set_data_to_obj( + file_data, + self.identifier, + key=key, + ) + + def _load_all_envs(self, source_data, silent=True, key=None): + """Load configs from files separating by each environment""" + + for file_data in source_data.values(): + + # env name is checked in lower + file_data = {k.lower(): value for k, value in file_data.items()} + + # is there a `dynaconf_merge` on top level of file? + file_merge = file_data.get("dynaconf_merge") + + # is there a flag disabling dotted lookup on file? + file_dotted_lookup = file_data.get("dynaconf_dotted_lookup") + + for env in build_env_list(self.obj, self.env): + env = env.lower() # lower for better comparison + + try: + data = file_data[env] or {} + except KeyError: + if silent: + continue + raise + + if not data: + continue + + self._set_data_to_obj( + data, + f"{self.identifier}_{env}", + file_merge, + key, + file_dotted_lookup=file_dotted_lookup, + ) + + def _set_data_to_obj( + self, + data, + identifier, + file_merge=None, + key=False, + file_dotted_lookup=None, + ): + """Calls settings.set to add the keys""" + # data 1st level keys should be transformed to upper case. + data = {upperfy(k): v for k, v in data.items()} + if key: + key = upperfy(key) + + if self.obj.filter_strategy: + data = self.obj.filter_strategy(data) + + # is there a `dynaconf_merge` inside an `[env]`? + file_merge = file_merge or data.pop("DYNACONF_MERGE", False) + + # If not passed or passed as None, + # look for inner [env] value, or default settings. + if file_dotted_lookup is None: + file_dotted_lookup = data.pop( + "DYNACONF_DOTTED_LOOKUP", + self.obj.get("DOTTED_LOOKUP_FOR_DYNACONF"), + ) + + if not key: + self.obj.update( + data, + loader_identifier=identifier, + merge=file_merge, + dotted_lookup=file_dotted_lookup, + ) + elif key in data: + self.obj.set( + key, + data.get(key), + loader_identifier=identifier, + merge=file_merge, + dotted_lookup=file_dotted_lookup, + ) |