"""Extends the import capability of YAML files.
"""
__author__ = 'Paul Landes'
from typing import Dict, Any, Sequence, List, Set, Union, ClassVar
import logging
import random
import string
from . import (
ConfigurableError, Serializer,
Configurable, DictionaryConfig, ImportIniConfig
)
logger = logging.getLogger(__name__)
[docs]
class ImportTreeConfig(DictionaryConfig):
"""A :class:`.Configurable` that give :class:`.ImortIniConfig` capabilities
to any gree based configuration, and specifically
:class:`.ImportYamlConfig`.
"""
DEFAULT_TYPE_MAP: ClassVar[Dict[str, str]] = {
'yml': 'condyaml', 'conf': 'importini'}
[docs]
def __init__(self, parent_section_name: str, parent: Configurable,
**kwargs):
self._parent_section_name = parent_section_name
self._import_section = None
super().__init__(default_section=None, parent=parent)
self._import_section: Dict[str, Any] = kwargs
def _get_config(self) -> Dict[str, Any]:
config: Dict[str, Any] = super()._get_config()
if self._import_section is not None:
self._import_tree(self._import_section, config)
self._import_section = None
return config
def _create_sec_name(self) -> str:
"""Return a unique section name based off this (parent) section name."""
sec_name: str = self._parent_section_name
if sec_name is None or len(sec_name) == 0:
sec_name = ''
rand = ''.join(random.choices(string.ascii_lowercase, k=10))
return f'{sec_name}_{rand}'
def _create_import_config(self, sec: Dict[str, Any]) -> Dict[str, Any]:
"""Return a dict with the :class:`.Configurable` import sections and a
list of cleanups.
"""
def map_cf(obj) -> str:
if isinstance(obj, Dict) and len(obj) == 1:
t = next(iter(obj.items()))
return f'{t[0]}: {t[1]}'
return obj
ser: Serializer = self.serializer
# create a unique import section name
sec_name: str = self._create_sec_name()
# sections to remove after
cleanups: List[str] = sec.pop(ImportIniConfig.CLEANUPS_NAME, [])
# the extension to Configurable class
type_map: str = sec.pop(ImportIniConfig.TYPE_MAP, self.DEFAULT_TYPE_MAP)
# get the config files to load (plural and singular nomenclatures)
files: Union[str, Sequence[Any]] = sec.get(
ImportIniConfig.CONFIG_FILES, [])
sfile: str = sec.pop(ImportIniConfig.SINGLE_CONFIG_FILE, None)
# find entries that don't belong
unknown_entries: Set[str] = \
(set(sec.keys()) - ImportIniConfig.IMPORT_SECTION_FIELDS)
if len(unknown_entries) > 0:
raise ConfigurableError(
f'Unknown configuration entries: {unknown_entries}')
# adding files/resources to load can be as a string, list, or what the
# YAML parser creates as a list of dicts for ``resource:...``; validate
if isinstance(files, str):
files = tuple(filter(lambda s: len(s) > 0, files.split('\n')))
if sfile is not None:
files.append(ser.format_option(sfile))
files = tuple(map(map_cf, files))
if len(files) == 0:
raise ConfigurableError(f'No configuration files set: {self}')
# add the processed files to the inmport section
sec[ImportIniConfig.CONFIG_FILES] = ser.format_option(files)
# and the current section (yaml based) import section
if self._parent_section_name is not None:
cleanups.append(self._parent_section_name)
# remove ``type: treeimport`` from the section
sec.pop(ImportIniConfig.TYPE_NAME, None)
# create the import section
imp_sec: Dict[str, Any] = {
ImportIniConfig.SECTIONS_SECTION:
ser.format_option([sec_name])}
# use the serializer to convert Python objects and types to strings
op: str
for op in (ImportIniConfig.REFS_NAME, ImportIniConfig.CLEANUPS_NAME):
val: Any = sec.pop(op, None)
if val is not None:
imp_sec[op] = ser.format_option(val)
# same for the top level ``import`` section
for op in (ImportIniConfig.ENABLED_NAME,):
val: Any = sec.pop(op, None)
if val is not None:
sec[op] = ser.format_option(val)
sec[ImportIniConfig.TYPE_MAP] = ser.format_option(type_map)
# create the config with the top level ``import`` and import section
config = DictionaryConfig({ImportIniConfig.IMPORT_SECTION: imp_sec,
sec_name: sec})
return {'config': config,
'cleanups': cleanups}
def _import_tree(self, source: Dict[str, Any], target: Dict[str, Any]):
"""Create a new :class:`.ImportIniConfig` with import sections to load
configuration for the section build for this instance.
:param source: the section in the YAML that tells us what to import
:param target: the data to populate with the imported data
"""
# create the input to ImportIniConfig
import_data: Dict[str, Any] = self._create_import_config(source)
cleanups: List[str] = import_data['cleanups']
# create the import configurable instance with importini context
iniconfig = ImportIniConfig(
import_data['config'],
children=self.parent.children,
parent=self.parent)
# preemptively remove sections so they deep dicts don't interfere with
# the copying of sections
if len(cleanups) > 0:
if logger.isEnabledFor(logging.DEBUG):
logger.debug(f'removing {cleanups}')
parent: Configurable = self.parent
while parent is not None:
if parent._is_initialized():
for sec in cleanups:
parent.remove_section(sec)
parent = parent.parent
# copy the imported sections to our config
dc = DictionaryConfig(target)
iniconfig.copy_sections(dc)