"""YAML configuration importation like :class:`.ImportIniConfig`.
"""
__author__ = 'Paul Landes'
from typing import Dict, Union, Any, Set, Tuple, ClassVar
from pathlib import Path
import logging
import re
from string import Template
from io import TextIOBase
from . import (
ConfigurableError, Serializer, Configurable, ConfigurableFactory,
DictionaryConfig, YamlConfig,
)
logger = logging.getLogger(__name__)
class _Template(Template):
idpattern = r'[a-z0-9_:]+'
def _dict_merge(target: Dict[Any, Any], source: Dict[Any, Any]):
""" Recursive dict merge. Inspired by :meth:``dict.update()``, instead of
updating only top-level keys, dict_merge recurses down into dicts nested
to an arbitrary depth, updating keys. The ``source`` is merged into
``target``.
:param target: dict onto which the merge is executed
:param source: merged into ``target``
:see: `Attribution <https://gist.github.com/angstwad/bf22d1822c38a92ec0a9>`_
"""
for k, v in source.items():
if (k in target and isinstance(target[k], dict) and \
isinstance(source[k], dict)): # noqa
_dict_merge(target[k], source[k])
else:
target[k] = source[k]
[docs]
class ImportYamlConfig(YamlConfig):
"""Like :class:`.YamlConfig` but supports configuration importation like
:class:`.ImportIniConfig`. The list of imports is given at
:obj:`import_name` (see initializer), and contains the same information as
import sections documented in :class:`.ImportIniConfig`.
"""
_KEY_PAT: ClassVar[re.Pattern] = re.compile(r'^\$\{(?P<key>[a-z0-9_:]+)\}$')
[docs]
def __init__(self, config_file: Union[Path, TextIOBase] = None,
default_section: str = None, sections_name: str = 'sections',
sections: Set[str] = None, import_name: str = 'import',
parse_values: bool = False,
children: Tuple[Configurable, ...] = (),
parent: Configurable = None,
merge_strategy_name: str = 'merge',
default_merge_strategy: str = 'parent'):
"""Initialize with importation configuration. The usage of
``default_vars`` in the super class is disabled since this
implementation uses a mix of dot and colon (configparser) variable
substitution (the later used when imported from an
:class:`.ImportIniConfig`.
:param config_file: the configuration file path to read from; if the
type is an instance of :class:`io.TextIOBase`, then
read it as a file object
:param default_section: used as the default section when non given on
the get methds such as :meth:`get_option`;
which defaults to ``defualt``
:param sections_name: the dot notated path to the variable that has a
list of sections
:param sections: used as the set of sections for this instance
:param import_name: the dot notated path to the variable that has the
import entries (see class docs); defaults to
``import``
:param parse_values: whether to invoke the :class:`.Serializer` to
create in memory Python data values, which defaults
to false to keep data as strings for configuration
merging
:param merge_strategy_name: property name in the import section to
indicate how to merge imported sections
:param default_merge_strategy: merge strategy if not given
"""
super().__init__(config_file, default_section, parent=parent,
default_vars=None, delimiter=None,
sections_name=sections_name, sections=sections)
self.import_name = import_name
self.serializer = Serializer()
self._parse_values = parse_values
self.children = children
self.merge_strategy_name = merge_strategy_name
self.default_merge_strategy = default_merge_strategy
def _interpolate(self, val: str, context: dict[str, object]):
repl: object = None
m: re.Match = self._KEY_PAT.match(val)
if m is not None:
key: str = m.group(1)
cval: object = context.get(key)
if cval is not None:
repl = cval
if repl is None:
template = _Template(val)
repl = template.safe_substitute(context)
return repl
def _merge_section(self, cnf: Dict[str, Any], section_name: str,
child: Configurable, strategy: str):
child_section: Dict[str, Any] = child.get_options(section_name)
if logger.isEnabledFor(logging.DEBUG):
logger.debug(f'merge: {child_section}')
if strategy == 'replace':
cnf[section_name] = child_section
elif strategy in {'child', 'parent'}:
parent_section: Dict[str, Any] = cnf.get(section_name)
if parent_section is None:
parent_section = {}
if strategy == 'parent':
_dict_merge(parent_section, child_section)
cnf[section_name] = parent_section
else:
_dict_merge(child_section, parent_section)
cnf[section_name] = child_section
else:
raise ConfigurableError(
f'Unknown YAML section merge strategy: {strategy}, '
"must be one of 'replace', 'child', 'parent'")
def _import_parse(self):
def repl_node(par: Dict[str, Any]):
repl = {}
for k, c in par.items():
if isinstance(c, dict):
repl_node(c)
elif isinstance(c, list):
repl[k] = tuple(c)
elif isinstance(c, str):
rc = self._interpolate(c, tpl_context)
if logger.isEnabledFor(logging.DEBUG):
logger.debug(f'subs: {k} = {c} -> {rc}')
repl[k] = rc
par.update(repl)
import_def: Dict[str, Any] = self.get_options(
f'{self.root}.{self.import_name}')
cnf: Dict[str, Any] = self._config
context: Dict[str, str] = {}
tpl_context = {}
if logger.isEnabledFor(logging.DEBUG):
logger.debug(f'import defs: {import_def}')
if import_def is not None:
sec_name: str
params: Dict[str, Any]
for sec_name, params in import_def.items():
if logger.isEnabledFor(logging.DEBUG):
logger.debug(f'import sec: {sec_name}')
strategy = params.pop(
self.merge_strategy_name, self.default_merge_strategy)
config: Configurable = ConfigurableFactory.from_section(
params, sec_name, parent=self)
child_sec: str
for child_sec in config.sections:
self._merge_section(cnf, child_sec, config, strategy)
if logger.isEnabledFor(logging.DEBUG):
logger.debug(f'updated config: {self._config}')
self._flatten(context, '', self._config, ':')
if len(self.children) > 0:
dconf = DictionaryConfig(parent=self)
for child in self.children:
child.copy_sections(dconf)
tpl_context.update(dconf.as_one_tier_dict())
tpl_context.update(context)
new_keys = set(map(lambda k: k.replace(':', '.'), context.keys()))
self._all_keys.update(new_keys)
repl_node(self._config)
def _serialize(self, par: Dict[str, Any]):
repl = {}
for k, c in par.items():
if isinstance(c, dict):
self._serialize(c)
elif isinstance(c, str):
repl[k] = self.serializer.parse_object(c)
par.update(repl)
def _compile(self) -> Dict[str, Any]:
self._config = super()._compile()
self._import_parse()
if self._parse_values:
self._serialize(self._config)
return self._config