Source code for zensols.config.strconfig
"""Implementation classes that are used as application configuration containers
parsed from files.
"""
__author__ = 'Paul Landes'
from typing import Dict, Set, Union
import logging
import re
import collections
from zensols.persist import persisted
from . import Configurable, ConfigurableError
logger = logging.getLogger(__name__)
[docs]
class StringConfig(Configurable):
"""A simple string based configuration. This takes a single comma delimited
key/value pair string in the format:
``<section>.<name>=<value>[,<section>.<name>=<value>,...]``
A dot (``.``) is used to separate the section from the option instead of a
colon (``:``), as used in more sophisticaed interpolation in the
:class:`configparser.ExtendedInterpolation`. The dot is used for this
reason to make other section interpolation easier.
"""
KEY_VAL_REGEX = re.compile(r'^(?:([^.]+?)\.)?([^=]+?)=(.+)$')
[docs]
def __init__(self, config_str: str,
option_sep_regex: Union[re.Pattern, str] = r'\s*,\s*',
default_section: str = None, parent: Configurable = None):
"""Initialize with a string given as described in the class docs.
:param config_str: the configuration
:param option_sep_regex: the regular expression used to delimit the each
key/value pair
:param default_section: used as the default section when non given on
the get methds such as :meth:`get_option`
"""
super().__init__(default_section, parent=parent)
self.config_str = config_str
if isinstance(option_sep_regex, str):
self.option_sep_regex = re.compile(option_sep_regex)
else:
self.option_sep_regex = option_sep_regex
self._initialized = False
@persisted('_parsed_config')
def _get_parsed_config(self) -> Dict[str, str]:
"""Parse the configuration string given in the initializer (see class
docs).
"""
conf = collections.defaultdict(lambda: {})
for kv in re.split(self.option_sep_regex, self.config_str):
m = self.KEY_VAL_REGEX.match(kv)
if m is None:
raise ConfigurableError(
f"unexpected format: '{kv}' in '{self.config_str}'")
sec, name, value = m.groups()
sec = self.default_section if sec is None else sec
if logger.isEnabledFor(logging.DEBUG):
logger.debug(f'section={sec}, name={name}, value={value}')
conf[sec][name] = value
self._initialized = True
return conf
@property
@persisted('_sections')
def sections(self) -> Set[str]:
return set(self._get_parsed_config().keys())
[docs]
def has_option(self, name: str, section: str = None) -> bool:
section = self.default_section if section is None else section
return self._get_parsed_config(section)[name]
[docs]
def get_options(self, section: str = None) -> Dict[str, str]:
section = self.default_section if section is None else section
opts = self._get_parsed_config()[section]
if opts is None:
raise ConfigurableError(f'no section: {section}')
return opts
def _is_initialized(self) -> bool:
return self._initialized
def __str__(self) -> str:
return self.__class__.__name__ + ': config=' + self.config_str
def __repr__(self) -> str:
return f'<{self.__str__()}>'