"""
This module contains the settings loaders provided by Typed Settings and the
protocol specification that they must implement.
"""
import importlib.util
import logging
import os
from collections.abc import Iterable
from fnmatch import fnmatch
from pathlib import Path
from typing import (
Any,
Callable,
Optional,
Protocol,
Union,
cast,
)
from ._compat import PY_311
if PY_311:
import tomllib
else:
import tomli as tomllib # type: ignore[no-redef]
from . import cls_utils
from .dict_utils import set_path
from .exceptions import (
ConfigFileLoadError,
ConfigFileNotFoundError,
InvalidOptionsError,
UnknownFormatError,
)
from .types import (
LoadedSettings,
LoaderMeta,
OptionInfo,
OptionList,
SettingsClass,
SettingsDict,
)
__all__ = [
"DictLoader",
"EnvLoader",
"FileFormat",
"FileLoader",
"InstanceLoader",
"Loader",
"OnePasswordLoader",
"PythonFormat",
"TomlFormat",
"clean_settings",
]
LOGGER = logging.getLogger("typed_settings")
[docs]
class Loader(Protocol):
"""
**Protocol** that settings loaders must implement.
Loaders must be callables (e.g., functions) with the specified signature.
.. versionchanged:: 1.0.0
Renamed ``load()`` to ``__call__()`` and also pass the settings
class.
"""
[docs]
def __call__(
self, settings_cls: SettingsClass, options: OptionList
) -> Union[LoadedSettings, Iterable[LoadedSettings]]:
"""
Load settings for the given options.
Args:
settings_cls: The base settings class for all options.
options: The list of available settings.
Return:
A dict with the loaded settings.
"""
...
class _DefaultsLoader:
"""
Return the default settings for a given settings class.
Args:
base_dir: Resolve relative paths in default settings relative to this directory.
By default, use the user's current working directory.
For tools like linters or test runners, you may want to use the root
directory of the app that the tool operates in.
"""
def __init__(self, base_dir: Path = Path()) -> None:
self.base_dir = base_dir
def __call__(
self, settings_cls: SettingsClass, options: OptionList
) -> LoadedSettings:
settings: SettingsDict = {}
# Populate dict with default settings. This avoids problems with nested
# settings classes for which no settings are loaded.
for opt in options:
if opt.has_no_default:
continue
if opt.default_is_factory:
# Do not invoke default factories yet. This should be done as late as
# possible. This is especially required for CLIs if you want to invoke
# the same instance multiple times (e.g., in tests).
continue
set_path(settings, opt.path, opt.default)
return LoadedSettings(settings, LoaderMeta(self, base_dir=self.base_dir))
[docs]
class DictLoader:
"""
Load settings from a dict of values.
This is mainly for testing purposes.
Args:
settings: A (nested) dict of settings
.. versionadded:: 23.1.0
"""
def __init__(self, settings: dict) -> None:
self.settings = settings
[docs]
def __call__(
self, settings_cls: SettingsClass, options: OptionList
) -> LoadedSettings:
"""
Load settings for the given options.
Args:
options: The list of available settings.
settings_cls: The base settings class for all options.
Return:
A dict with the loaded settings.
"""
return LoadedSettings(self.settings, LoaderMeta(self))
[docs]
class InstanceLoader:
"""
Load settings from an instance of the settings class.
Args:
instance: The settings instance from which to load option values.
.. versionadded:: 1.0.0
"""
def __init__(self, instance: object) -> None:
self.instance = instance
[docs]
def __call__(
self, settings_cls: SettingsClass, options: OptionList
) -> LoadedSettings:
"""
Load settings for the given options.
Args:
options: The list of available settings.
settings_cls: The base settings class for all options.
Return:
A dict with the loaded settings.
"""
if not isinstance(self.instance, settings_cls):
raise ValueError(
f'"self.instance" is not an instance of {settings_cls}: '
f"{type(self.instance)}"
)
cls_handler = cls_utils.find_handler(type(self.instance))
return LoadedSettings(cls_handler.asdict(self.instance), LoaderMeta(self))
[docs]
class EnvLoader:
"""
Load settings from environment variables.
Args:
prefix: Prefix for environment variables, e.g., ``MYAPP_``.
nested_delimiter: Delimiter for attribute names of nested classes.
"""
def __init__(self, prefix: str, nested_delimiter: str = "_") -> None:
self.prefix = prefix
self.nested_delimiter = nested_delimiter
[docs]
def __call__(
self, settings_cls: SettingsClass, options: OptionList
) -> LoadedSettings:
"""
Load settings for the given options.
Args:
options: The list of available settings.
settings_cls: The base settings class for all options.
Return:
A dict with the loaded settings.
"""
LOGGER.debug(f"Looking for env vars with prefix: {self.prefix}")
env = os.environ
values: SettingsDict = {}
for o in options:
varname = self.get_envvar(o)
if varname in env:
LOGGER.debug(f"Env var found: {varname}")
set_path(values, o.path, env[varname])
else:
LOGGER.debug(f"Env var not found: {varname}")
return LoadedSettings(values, LoaderMeta(self))
[docs]
def get_envvar(self, option: OptionInfo) -> str:
"""
Return the envvar name for the he given option.
"""
return f"{self.prefix}{option.path.upper().replace('.', self.nested_delimiter)}"
[docs]
class FileLoader:
"""
Load settings from config files.
Settings of multiple files will be merged. The last file has the highest
precedence. Files specified via an environment variable are loaded after
the files passed to this class, i.e.:
- First file from *files*
- ...
- Last file from *files*
- First file from *env_var*
- ...
- Last file from *env_var*
Mandatory files can be prefixed with ``!``. Optional files will be ignored
if they don't exist.
Args:
formats: A dict mapping glob patterns to :class:`FileFormat` instances.
files: A list of filenames to try to load.
env_var: Name of the environment variable that may hold additional file
paths. If it is ``None``, only files from *files* will be loaded.
"""
def __init__(
self,
formats: dict[str, FileFormat],
files: Iterable[Union[str, Path]],
env_var: Optional[str] = None,
) -> None:
self.files = files
self.env_var = env_var
self.formats = formats
[docs]
def __call__(
self, settings_cls: SettingsClass, options: OptionList
) -> list[LoadedSettings]:
"""
Load settings for the given options.
Args:
options: The list of available settings.
settings_cls: The base settings class for all options.
Return:
A dict with the loaded settings.
Raise:
UnknownFormatError: When no :class:`FileFormat` is configured for a
loaded file.
ConfigFileNotFoundError: If *path* does not exist.
ConfigFileLoadError: If *path* cannot be read/loaded/decoded.
InvalidOptionsError: If invalid settings have been found.
"""
paths = self._get_config_filenames(self.files, self.env_var)
loaded_settings: list[LoadedSettings] = []
for path in paths:
settings = self._load_file(path, settings_cls, options)
meta = LoaderMeta(f"{type(self).__name__}[{path}]", base_dir=path.parent)
loaded_settings.append(LoadedSettings(settings, meta))
return loaded_settings
def _load_file(
self,
path: Path,
settings_cls: SettingsClass,
options: OptionList,
) -> SettingsDict:
"""
Load a file and return its cleaned contents.
"""
# "clean_settings()" must be called for each loaded file individually
# because of the "-"/"_" normalization. This also allows us to tell
# the user the exact file that contains errors.
for pattern, ffloader in self.formats.items():
if fnmatch(path.name, pattern):
settings = ffloader(path, settings_cls, options)
settings = clean_settings(settings, options, path)
return settings
raise UnknownFormatError(f"No loader configured for: {path}")
@staticmethod
def _get_config_filenames(
files: Iterable[Union[str, Path]], env_var: Optional[str]
) -> list[Path]:
"""
Concatenate *config_files* and files from env var *config_files_var*.
"""
candidates = [(False, str(f)) for f in files]
if env_var:
LOGGER.debug(f"Env var for config files: {env_var}")
candidates += [(True, fname) for fname in os.getenv(env_var, "").split(":")]
else:
LOGGER.debug("Env var for config files not set")
paths = []
for from_envvar, fname in candidates:
_, flag, fname = fname.rpartition("!")
if not fname:
continue
is_mandatory = flag == "!"
try:
path = Path(fname).resolve(strict=True)
except FileNotFoundError:
if is_mandatory:
LOGGER.error(f"Mandatory config file not found: {fname}")
raise
if from_envvar:
LOGGER.warning(f"Config file from {env_var} not found: {fname}")
else:
LOGGER.info(f"Config file not found: {fname}")
else:
LOGGER.debug(f"Loading settings from: {path}")
paths.append(path)
return paths
[docs]
class OnePasswordLoader:
"""
Load settings from an item stored in a 1Password vault.
You must must have installed and set up the `1Password CLI`_ in order
for this loader to work.
.. _1Password CLI: https://developer.1password.com/docs/cli/
Args:
item: The item to load
vault: The vault in which to look for *item*. By default, search all
vaults.
"""
def __init__(self, item: str, vault: Optional[str] = None) -> None:
self.item = item
self.vault = vault
from . import _onepassword
self._op = _onepassword
[docs]
def __call__(self, settings_cls: type, options: OptionList) -> LoadedSettings:
"""
Load settings for the given options.
Args:
options: The list of available settings.
settings_cls: The base settings class for all options.
Return:
A dict with the loaded settings.
"""
option_names = [o.path for o in options]
settings = self._op.get_item(self.item, self.vault)
settings = {k: v for k, v in settings.items() if k in option_names}
return LoadedSettings(settings, LoaderMeta(self))
[docs]
def clean_settings(
settings: SettingsDict, options: OptionList, source: Any
) -> SettingsDict:
"""
Recursively check settings for invalid entries and raise an error.
An error is not raised until all options have been checked. It then lists
all invalid options that have been found.
Args:
settings: The settings to be cleaned.
options: The list of available settings.
source: Source of the settings (e.g., path to a config file).
It should have a useful string representation.
Return:
The cleaned settings.
Raise:
InvalidOptionsError: If invalid settings have been found.
"""
invalid_paths = []
valid_paths = {o.path for o in options}
cleaned: SettingsDict = {}
def _iter_dict(d: SettingsDict, prefix: str) -> None:
for key, val in d.items():
key = key.replace("-", "_")
path = f"{prefix}{key}"
if path in valid_paths:
set_path(cleaned, path, val)
continue
if isinstance(val, dict):
_iter_dict(val, f"{path}.")
else:
invalid_paths.append(path)
_iter_dict(settings, "")
if invalid_paths:
joined_paths = ", ".join(sorted(invalid_paths))
raise InvalidOptionsError(f"Invalid options found in {source}: {joined_paths}")
return cleaned