bcd877266f
The env loader is now capable of loading lists of objects, union types or list of union types from the env variables. They are some limitations: for example it doesn't support unions of different shapes `list | dict` or `str | dict`.
102 lines
3 KiB
Python
102 lines
3 KiB
Python
import sys
|
|
from itertools import zip_longest
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional, Union
|
|
|
|
from loguru import logger
|
|
from pydantic import BaseModel, ValidationError
|
|
from yaml import YAMLError, safe_load
|
|
|
|
from ._env import EnvLoader
|
|
|
|
DEFAULT_ENV_PREFIX = "LIBRETIME"
|
|
DEFAULT_CONFIG_FILEPATH = Path("/etc/libretime/config.yml")
|
|
|
|
|
|
# pylint: disable=too-few-public-methods
|
|
class BaseConfig(BaseModel):
|
|
"""
|
|
Read and validate the configuration from 'filepath' and os environment.
|
|
|
|
:param filepath: yaml configuration file to read from
|
|
:param env_prefix: prefix for the environment variable names
|
|
:param env_delimiter: delimiter for the environment variable names
|
|
:returns: configuration class
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
env_prefix: str = DEFAULT_ENV_PREFIX,
|
|
env_delimiter: str = "_",
|
|
filepath: Optional[Union[Path, str]] = None,
|
|
) -> None:
|
|
if filepath is not None:
|
|
filepath = Path(filepath)
|
|
|
|
env_loader = EnvLoader(self.schema(), env_prefix, env_delimiter)
|
|
|
|
values = deep_merge_dict(
|
|
self._load_file_values(filepath),
|
|
env_loader.load(),
|
|
)
|
|
|
|
try:
|
|
super().__init__(**values)
|
|
except ValidationError as error:
|
|
logger.critical(error)
|
|
sys.exit(1)
|
|
|
|
def _load_file_values(
|
|
self,
|
|
filepath: Optional[Path] = None,
|
|
) -> Dict[str, Any]:
|
|
if filepath is None:
|
|
logger.debug("no config filepath is provided")
|
|
return {}
|
|
|
|
if not filepath.is_file():
|
|
logger.warning(f"provided config filepath '{filepath}' is not a file")
|
|
return {}
|
|
|
|
try:
|
|
return safe_load(filepath.read_text(encoding="utf-8"))
|
|
except YAMLError as error:
|
|
logger.error(f"config file '{filepath}' is not a valid yaml file: {error}")
|
|
|
|
return {}
|
|
|
|
|
|
def deep_merge_dict(base: Dict[str, Any], next_: Dict[str, Any]) -> Dict[str, Any]:
|
|
result = base.copy()
|
|
for key, value in next_.items():
|
|
if key in result:
|
|
if isinstance(result[key], dict) and isinstance(value, dict):
|
|
result[key] = deep_merge_dict(result[key], value)
|
|
continue
|
|
|
|
if isinstance(result[key], list) and isinstance(value, list):
|
|
result[key] = deep_merge_list(result[key], value)
|
|
continue
|
|
|
|
if value:
|
|
result[key] = value
|
|
|
|
return result
|
|
|
|
|
|
def deep_merge_list(base: List[Any], next_: List[Any]) -> List[Any]:
|
|
result: List[Any] = []
|
|
for base_item, next_item in zip_longest(base, next_):
|
|
if isinstance(base_item, list) and isinstance(next_item, list):
|
|
result.append(deep_merge_list(base_item, next_item))
|
|
continue
|
|
|
|
if isinstance(base_item, dict) and isinstance(next_item, dict):
|
|
result.append(deep_merge_dict(base_item, next_item))
|
|
continue
|
|
|
|
if next_item:
|
|
result.append(next_item)
|
|
|
|
return result
|