Source code for zensols.config.envconfig

"""An implementation configuration class that holds environment variables.

"""
__author__ = 'Paul Landes'

from typing import Dict, Set
import logging
import collections
import os
from zensols.persist import persisted
from . import ConfigurableError, Configurable

logger = logging.getLogger(__name__)


[docs] class EnvironmentConfig(Configurable): """An implementation configuration class that holds environment variables. This config will need to be added to children to :class:`.ImportIniConfig` if used in the configuration or import sections. """
[docs] def __init__(self, section_name: str = 'env', map_delimiter: str = None, skip_delimiter: bool = False, includes: Set[str] = None): """Initialize with a string given as described in the class docs. The string ``<DOLLAR>`` used with ``map_delimiter`` is the same as ``$`` since adding the dollar in some configuration scenarios has parsing issues. One example is when ``$$`` failes on copying section to an :class:`.IniConfig`. :param section_name: the name of the created section with the environment variables :param map_delimiter: when given, all environment values are replaced with a duplicate; set this to ``$`` when using :class:`configparser.ExtendedInterpolation` for environment variables such as ``PS1`` :param skip_delimiter: a string, when present, causes the environment variable to be skipped; this is useful for environment variables that cause interpolation errors :param includes: if given, the set of environment variables to set excluding the rest; include all if ``None`` """ super().__init__(section_name) if map_delimiter == '<DOLLAR>': map_delimiter = '$' self.map_delimiter = map_delimiter self.skip_delimiter = skip_delimiter self.includes = includes
@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 self.config_str.split(self.option_sep): m = self.KEY_VAL_REGEX.match(kv) if m is None: raise ConfigurableError(f'unexpected format: {kv}') 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 return conf @persisted('_keys') def _get_keys(self) -> Dict[str, str]: return self._get_parsed_config().keys() @property @persisted('_sections') def sections(self) -> Set[str]: return frozenset([self.default_section])
[docs] def has_option(self, name: str, section: str = None) -> bool: keys = self._get_keys() return self.default_section == section and name in keys
@persisted('_env_section') def _get_env_section(self) -> Dict[str, str]: opts = {} delim = self.map_delimiter if delim is not None: repl = f'{delim}{delim}' for k, v in os.environ.items(): if ((self.includes is not None) and (k not in self.includes)) or \ (self.skip_delimiter and v.find(delim) >= 0): continue if delim is None: val = v else: val = v.replace(delim, repl) opts[k] = val return opts
[docs] def get_options(self, section: str = None) -> Dict[str, str]: if section == self.default_section: opts = self._get_env_section() else: opts = {} return opts