"""Configuration merging first pass applications. These utility applications
allow a configuration option and then merge that configuration with the
application context.
"""
__author__ = 'Paul Landes'
from typing import Dict, Any, Set, List, Tuple, Iterable, Optional, Type, Union
from dataclasses import dataclass, field
import os
import logging
from string import Template
import parse as par
import re
from pathlib import Path
import pickle
from zensols.util import PackageResource
from zensols.persist import persisted
from zensols.config import (
rawconfig, Configurable, ConfigurableFactory,
IniConfig, ImportIniConfig, StringConfig, DictionaryConfig,
)
from .. import (
Dictable, ActionCliError, ApplicationError, OptionMetaData, ActionMetaData,
ApplicationObserver, Action, Application,
)
logger = logging.getLogger(__name__)
class _ConfiguratorImporterTemplate(Template):
delimiter = '^'
class _PreLoadImportIniConfig(ImportIniConfig):
_PRELOAD_FORMAT = 'preload:{}'
def __init__(self, *args, preloads: Dict[str, Configurable], **kwargs):
super().__init__(*args, **kwargs)
self.preloads = preloads
@classmethod
def format_preload(self, name: str) -> str:
return self._PRELOAD_FORMAT.format(name)
@classmethod
def parse_preload(self, s: str) -> Optional[str]:
pres: par.Result = par.parse(self._PRELOAD_FORMAT, s)
if pres is not None:
return pres[0]
def _create_config(self, section: str,
params: Dict[str, Any]) -> Configurable:
conf: Configurable = None
key: str = None
config_file: Union[Path, str] = params.get(self.SINGLE_CONFIG_FILE)
if isinstance(config_file, str):
key = self.parse_preload(config_file)
if key is not None:
conf = self.preloads.get(key)
if logger.isEnabledFor(logging.DEBUG):
logger.debug(f"create config '{key} -> {conf} ({type(conf)})")
if conf is None:
logger.warning(f"matched a preload format '{config_file}' " +
f"with non-existing preload: '{key}'")
if conf is None:
conf = super()._create_config(section, params)
return conf
[docs]
@dataclass
class ConfigurationOverrider(object):
"""Overrides configuration in the app config. This is useful for replacing
on a per command line invocation basis. Examples could include changing the
number of epochs trained for a model.
The :obj:`override` field either contains a path to a file that contains
the configuration file to use to clobber the given sections/values, or a
string to be interpreted by :class:`.StringConfig`. This determination is
made by whether or not the string points to an existing file or directory.
"""
OVERRIDE_FIELD = 'override'
CLI_META = {'first_pass': True, # not a separate action
'mnemonic_includes': {'merge'},
# better/shorter long name, and reserve the short name
'option_overrides': {OVERRIDE_FIELD:
{'metavar': '<FILE|DIR|STRING>',
'short_name': None}},
# only the path to the configuration should be exposed as a
# an option on the comamnd line
'option_includes': {OVERRIDE_FIELD}}
config: Configurable = field()
"""The parent configuration, which is populated from the child configuration
(see class docs).
"""
override: str = field(default=None)
"""A config file/dir or a comma delimited section.key=value string that
overrides configuration.
"""
option_sep: str = field(default=',')
"""The string used to delimit the each key/value pair."""
disable: bool = field(default=False)
"""Whether to disable the application, which is useful to set to ``False``
when used with :class:`.ConfigurationImporter`.
"""
[docs]
def get_configurable(self) -> Optional[Configurable]:
if self.override is not None:
path = Path(self.override)
if path.exists():
cf = ConfigurableFactory()
overrides = cf.from_path(path)
else:
overrides = StringConfig(
self.override,
option_sep=self.option_sep
)
return overrides
[docs]
def merge(self) -> Configurable:
"""Merge the string configuration with the application context."""
if logger.isEnabledFor(logging.DEBUG):
logger.debug(f'overriding with: {self.override}')
if not self.disable:
conf: Optional[Configurable] = self.get_configurable()
if conf is not None:
self.config.merge(conf)
return self.config
def __call__(self) -> Configurable:
return self.merge()
@dataclass
class _CacheConfigManager(object):
path: Path = field()
config: ImportIniConfig = field()
watch_files: Dict[Path, int] = field(default_factory=dict)
def _get_newer_files(self, files: Dict[Path, int]) -> Iterable[Path]:
path: Path
prev_mtime: int
for path, prev_mtime in files.items():
cur_mtime = path.stat().st_mtime
if cur_mtime > prev_mtime:
yield path
def load(self) -> ImportIniConfig:
prev_config: ImportIniConfig = None
if self.path.is_file():
with open(self.path, 'rb') as f:
prev_mng: _CacheConfigManager = pickle.load(f)
newer_files: Tuple[Path, ...] = tuple(
self._get_newer_files(prev_mng.watch_files))
if len(newer_files) == 0:
if logger.isEnabledFor(logging.INFO):
logger.info(f'reusing cached config: {self.path}')
prev_config = prev_mng.config
else:
if logger.isEnabledFor(logging.INFO):
fnames: str = ', '.join(map(str, newer_files))
logger.info(f'reloading since files changed: {fnames}')
if prev_config is None:
self.config.start_file_capture()
return prev_config
def save(self):
visited: Set[Path] = set(self.config.stop_file_capture())
self.watch_files.update(dict(map(
lambda p: (p, p.stat().st_mtime), visited)))
self.path.parent.mkdir(parents=True, exist_ok=True)
with open(self.path, 'wb') as f:
pickle.dump(self, f)
[docs]
@dataclass
class ConfigurationImporter(ApplicationObserver, Dictable):
"""This class imports a child configuration in to the application context.
It does this by:
1. Attempt to load the configuration indicated by the ``--config``
option.
2. If the option doesn't exist, attempt to get the path to load from an
environment variable (see :meth:`get_environ_var_from_app`). For
example, for package ``zensols.util``, the environment variable
``UTILRC`` environment variable's path is used.
3. If the environment variable is not found, then look for the UNIX style
resource file (see :meth:`get_environ_path`). For example, package
``zensols.util`` would point to ``~/.utilrc``.
4. Loads the *child* configuration.
5. Copy all sections from the child configuration to :obj:`config`.
The child configuration is created by :class:`.ConfigurableFactory`. If
the child has a `.conf` extension, :class:`.ImportIniConfig` is used with
its child set as :obj:`config` so the two can reference each other at
property/factory resolve time.
Special mnemonic ``^{config_path}`` can be used in an
:class`.ImportIniConfig` import section in the `config_files` property to
load the referred configuration file in any order with the other loaded
files. The special mnemonic ``^{override}`` does the same thing with the
:class:`.ConfigurationOverrider` one pass application as well.
"""
_OVERRIDES_KEY = ConfigurationOverrider.OVERRIDE_FIELD
_OVERRIDES_PRELOAD = _PreLoadImportIniConfig.format_preload(_OVERRIDES_KEY)
CONFIG_PATH_FIELD = 'config_path'
"""The field name in this class of the child configuration path.
:see: :obj:`config_path`
"""
CLI_META = {'first_pass': True, # not a separate action
# the mnemonic must be unique and used to referece the method
'mnemonic_overrides': {'merge': '_merge_config_as_import'},
'mnemonic_includes': {'merge'},
# better/shorter long name, and reserve the short name
'option_overrides': {CONFIG_PATH_FIELD: {'long_name': 'config',
'short_name': 'c'}},
# only the path to the configuration should be exposed as a
# an option on the comamnd line
'option_includes': {CONFIG_PATH_FIELD}}
"""Command line meta data to avoid having to decorate this class in the
configuration. Given the complexity of this class, this configuration only
exposes the parts of this class necessary for the CLI.
"""
IMPORT_TYPE = 'import'
"""The string value of the ``type`` parameter in the section config
identifying an :class:`.ImportIniConfig` import section.
"""
ENVIRON_VAR_REGEX = re.compile(r'^.+\.([a-z]+?)$')
"""A regular expression to parse the name from the package name for the
environment variable that might hold the configuration
(i.e. ``APPNAMERC``).
"""
name: str = field()
"""The section name."""
config: Configurable = field()
"""The parent configuration, which is populated from the child configuration
(see class docs).
"""
expect: bool = field(default=True)
"""If ``True``, raise an :class:`.ApplicationError` if the option is not
given.
"""
default: Path = field(default=None)
"""Use this file as the default when given on the command line, which is not
used unless :obj:``expect`` is set to ``False``.
If this is set to ``skip``, then do not load any file. This is useful when
the entire configuration is loaded by this class and there are
configuration mentions in the ``app.conf`` application context.
"""
config_path_environ_name: str = field(default=None)
"""An environment variable containing the default path to the configuration.
"""
type: str = field(default=None)
"""The type of :class:`.Configurable` use to create in
:class:`.ConfigurableFactory`. If this is not provided, the factory
decides based on the file extension.
:see: :class:`.ConfigurableFactory`
"""
arguments: Dict[str, Any] = field(default=None)
"""Additional arguments to pass to the :class:`.ConfigFactory` when created.
"""
section: str = field(default=None)
"""Additional which section to load as an import. This is only valid and
used when :obj:`type` is set to `import`. When it is, the section will
replace the string ``^{config_apth}`` (and any other field in this instance
using the same syntax) and load indicated remaining configuration using
:class:`~zensols.config.ImportIniConfig`.
See the `API documentation
<https://plandes.github.io/util/doc/config.html#import-ini-configuration>`_
for more information.
"""
config_path_option_name: str = field(default='config_path')
"""If not ``None``, the name of the option to set in the section defined for
this instance (section = :obj:`name`).
"""
debug: bool = field(default=False)
"""Printn the configuration after the merge operation."""
cache_path: Path = field(default=None)
"""A path to a binary file of the cached configuration. This is useful for
large application contexts that have a lot of imports that take a long time
to load.
When this file is set, the imported configuration files and their modify
timestamps are written along with the loaded configuration to this specified
file. On subsequent runs, previous configuration is used if none of the
configuration files have changed. Otherwise, they are reloaded and cached
again.
"""
# name of this field must match
# :obj:`ConfigurationImporter.CONFIG_PATH_FIELD`
config_path: Path = field(default=None)
"""The configuration file."""
@persisted('_rc_prefix')
def _get_rc_prefix(self) -> str:
"""Return the resource prefix. For example, if the package resource is
``zensols.util`` the outupt is ``util``.
"""
pkg_res: PackageResource = self._app.factory.package_resource
name: str = pkg_res.name
if logger.isEnabledFor(logging.DEBUG):
logger.debug(f"match environment variable '{name}' " +
f'on {self.ENVIRON_VAR_REGEX}')
m = self.ENVIRON_VAR_REGEX.match(name)
if m is not None:
name = m.group(1)
return name
[docs]
def get_environ_var_from_app(self) -> str:
"""Return the environment variable based on the name of the application.
This returns the :obj:`config_path_environ_name` if set, otherwise, it
generates it based on the name returned from the packge + ``RC`` and
capitalizes it.
"""
name: str
if self.config_path_environ_name is not None:
name = self.config_path_environ_name
else:
name = f'{self._get_rc_prefix()}rc'.upper()
if logger.isEnabledFor(logging.DEBUG):
logger.debug(f"using environment variable '{name}'")
return name
[docs]
def get_environ_path(self) -> Optional[Path]:
"""Return the path to the resource configuration file. This first uses
:meth:`get_environ_var_from_app` to attempt to find an environment
variable with a reference to file, then looks for it using the UNIX
style file naming conventions (i.e. package ``zensols.util`` would point
to ``~/.utilrc``).
"""
env_var: str = self.get_environ_var_from_app()
env_var_path: str = os.environ.get(env_var)
rc_path: Path = None
if logger.isEnabledFor(logging.DEBUG):
logger.debug('loading config from environment ' +
f"varaibles '{env_var}' = {env_var_path}")
if env_var_path is None:
prefix: str = self._get_rc_prefix()
path = Path(f'~/.{prefix}rc').expanduser()
if logger.isEnabledFor(logging.DEBUG):
logger.debug(f'trying UNIX style resource file: {path}')
if path.is_file():
rc_path = path
else:
path = Path(env_var_path)
if path.exists():
rc_path = path
else:
logger.warning(f'Environment variable {env_var} set to ' +
f'non-existant path: {env_var_path}')
return rc_path
def _get_config_option(self) -> str:
"""Return the long option name (with dashes) as given on the command
line.
"""
ameta: ActionMetaData = self._action.meta_data
ometa: OptionMetaData = ameta.options_by_dest[self.CONFIG_PATH_FIELD]
return ometa.long_option
def _application_created(self, app: Application, action: Action):
"""In this call back, set the app and action for using in the invocation
:meth:`merge`.
"""
if logger.isEnabledFor(logging.DEBUG):
logger.debug(f'configurator created with {action}')
self._app = app
self._action = action
def _get_override_config(self) -> Configurable:
conf: Configurable = None
ac: Action
for ac in self._app.actions:
ctype: Type = ac.class_meta.class_type
if issubclass(ctype, ConfigurationOverrider):
opts: Dict[str, Any] = ac.command_action.options
sconf: str = opts.get(ConfigurationOverrider.OVERRIDE_FIELD)
params = {ConfigurationOverrider.OVERRIDE_FIELD: sconf}
co = ConfigurationOverrider(self.config, **params)
conf = co.get_configurable()
break
if conf is None:
conf = DictionaryConfig()
return conf
def _validate(self):
# the section attribute is only useful for ImportIniConfig imports
if self.type != self.IMPORT_TYPE and self.section is not None:
raise ActionCliError("Cannot have a 'section' entry " +
f"without type of '{self.IMPORT_TYPE}'")
def _populate_import_sections(self, config: Configurable) -> \
Tuple[Configurable, Set[str]]:
sec: Dict[str, str] = self.config.get_options(self.section)
secs: Dict[str, Dict[str, str]] = {}
populated_sec = {}
vals = self.asdict()
vals[self._OVERRIDES_KEY] = self._OVERRIDES_PRELOAD
for k, v in sec.items():
populated_sec[k] = v.format(**vals)
secs[ImportIniConfig.IMPORT_SECTION] = populated_sec
preload_keys: Set[str] = set()
if ImportIniConfig.SECTIONS_SECTION in populated_sec:
sub_secs: List[str] = self.config.serializer.parse_object(
self.config.get_option(
ImportIniConfig.SECTIONS_SECTION, self.section))
for sec in sub_secs:
repl_sec: Dict[str, str] = {}
secs[sec] = repl_sec
with rawconfig(config):
for k, v in config.get_options(sec).items():
tpl = _ConfiguratorImporterTemplate(v)
try:
vr = tpl.substitute(vals)
except KeyError as e:
raise ActionCliError(
f"Bad config load special key {e} in: '{v}'")
if logger.isEnabledFor(logging.DEBUG):
logger.debug(f'{sec}:{k}: {v} -> {vr}')
pls = self.config.serializer.parse_object(vr)
if isinstance(pls, (list, tuple)):
pls = map(
_PreLoadImportIniConfig.parse_preload, pls)
pls = filter(lambda x: x is not None, pls)
preload_keys.update(pls)
repl_sec[k] = vr
return DictionaryConfig(secs), preload_keys
def _load_configuration(self) -> Configurable:
"""Once we have the path and the class used to load the configuration,
create the instance and load it.
Special handling is required to make options forward *and* backward
propogate.
"""
if logger.isEnabledFor(logging.INFO):
logger.info('configurator loading section: ' +
f'{self.config.config_file}:[{self.name}]')
# the modified configuration that will returned
modified_config: Configurable = self.config
# sections created during this call to later be removed
secs_to_del: Set[str] = set()
# create the command line specified config
do_back_copy: bool = True
# config section sanity check
self._validate()
# create a configuration factory using the configuration file extension
if self.type is None:
args = {} if self.arguments is None else self.arguments
cf = ConfigurableFactory(kwargs=args)
if self.config_path is None:
# this happens when expect=False and no configuration is given
cl_config = DictionaryConfig()
else:
cl_config = cf.from_path(self.config_path)
logger.info(f'config loaded {self.config_path} as ' +
f'type {cl_config.__class__.__name__}')
# import using ImportIniConfig as a section
elif self.type == self.IMPORT_TYPE:
args: dict = {} if self.arguments is None else self.arguments
children: Tuple[Configurable, ...] = \
self._app.factory.children_configs
ini: IniConfig = IniConfig(self.config)
dconf: Configurable
preload_keys: Set[str]
dconf, preload_keys = self._populate_import_sections(ini)
preloads: Dict[str, Configurable] = {}
if self._OVERRIDES_KEY in preload_keys:
preloads[self._OVERRIDES_KEY] = self._get_override_config()
secs_to_del.update(dconf.sections)
dconf.copy_sections(ini)
if children is None:
# unit test cases will not have children configurables
children = ()
with rawconfig(ini):
cl_config = _PreLoadImportIniConfig(
preloads=preloads,
config_file=ini,
children=children,
**args)
with rawconfig(cl_config):
cl_config.copy_sections(self.config)
modified_config = cl_config
# remove sections that were removed
removed_secs: Set[str] = self.config.sections - cl_config.sections
if logger.isEnabledFor(logging.DEBUG):
logger.debug(f'removing additional sections: {removed_secs}')
secs_to_del.update(removed_secs)
# avoid the two way copy that happens later
do_back_copy = False
# otherwise, use the type to tell the configuraiton factory how to
# create it
else:
args = {'config_file': self.config_path}
if self.arguments is not None:
args.update(self.arguments)
cf = ConfigurableFactory(kwargs=args)
cl_config = cf.from_type(self.type)
logger.info(f'configurator loading {self.config_path} from ' +
f'type {self.type}')
# For non-import configs, first inject our app context (app.conf) to
# the command line specified configuration (--config) skipping sections
# that have missing options. Examples of those missing include
# cyclical dependencies such option references from our app context to
# the command line context.
if do_back_copy:
if logger.isEnabledFor(logging.DEBUG):
logger.debug(f'copying app config to {cl_config}')
with rawconfig(self.config):
self.config.copy_sections(cl_config)
# copy the command line config to our app context letting it barf with
# any missing properties this time
if logger.isEnabledFor(logging.DEBUG):
logger.debug(f'copying to app config: {cl_config}')
if do_back_copy:
with rawconfig(cl_config):
cl_config.copy_sections(self.config)
# if we imported, we created ImportIniConfig sections we need to remove
for sec in secs_to_del:
if logger.isEnabledFor(logging.DEBUG):
logger.debug(f'removing (added) section: {sec}')
self.config.remove_section(sec)
if self.debug:
print(self.config.get_raw_str())
return modified_config
def _load(self) -> Configurable:
"""Load the configuration and update the application context.
"""
modified_config: Configurable = None
cmng: _CacheConfigManager = None
loaded: bool = False
if self.cache_path is not None:
if isinstance(self.config, ImportIniConfig):
if logger.isEnabledFor(logging.DEBUG):
logger.debug(f'caching: to {self.cache_path}')
cmng = _CacheConfigManager(self.cache_path, config=self.config)
if self.config_path is not None:
# TODO: add entry point app.conf to watch files
cmng.watch_files[self.config_path] = \
self.config_path.stat().st_mtime
modified_config = cmng.load()
if modified_config is not None:
with rawconfig(modified_config):
modified_config.copy_sections(self.config)
modified_config = self.config
loaded = True
else:
logger.warning('configured to cache config at' +
f'{self.cache_path} but configuration ' +
f'is not ImportIniConfig: {type(self.config)}')
if modified_config is None:
modified_config = self._load_configuration()
if cmng is not None and not loaded:
cmng.save()
if self.config_path_option_name is not None:
val: str
if self.config_path is None:
val = 'None'
else:
val = f'path: {str(self.config_path)}'
self.config.set_option(self.config_path_option_name,
val, section=self.name)
return modified_config
def _reset(self):
"""Reset the Python logger configuration."""
root = logging.getLogger()
tuple(map(root.removeHandler, root.handlers[:]))
tuple(map(root.removeFilter, root.filters[:]))
[docs]
def merge(self) -> Configurable:
"""Merge configuration at path to the current configuration.
:param config_path: the path to the configuration file
"""
# the modified configuration that will returned
modified_config: Configurable = self.config
rc_path: Path = None
if self.config_path is None:
rc_path: Path = self.get_environ_path()
if rc_path is not None:
self.config_path = rc_path
elif self.default is not None:
if self.default == 'skip':
self.config_path = None
else:
self.config_path = self.default
if self.config_path is None:
if self.expect:
long_opt: str = self._get_config_option()
raise ApplicationError(f'Missing option {long_opt}')
else:
modified_config = self._load()
else:
modified_config = self._load()
return modified_config
def __call__(self) -> Configurable:
return self.merge()