"""Implementation of the JSON configurable.
"""
__author__ = 'Paul Landes'
from typing import Union, Dict, Any
from dataclasses import dataclass
import logging
from pathlib import Path
from io import TextIOBase
import json
from zensols.persist import persisted, PersistedWork
from . import (
ConfigurableError, ConfigurableFileNotFoundError, DictionaryConfig
)
logger = logging.getLogger(__name__)
[docs]
@dataclass
class JsonConfig(DictionaryConfig):
"""A configurator that reads JSON as a two level dictionary. The top level
keys are the section and the values are a single depth dictionary with
string keys and values.
A caveat is if all the values are terminal, in which case the top level
singleton section is ``default_section`` given in the initializer and the
section content is the single dictionary.
"""
[docs]
def __init__(self, config_file: Union[Path, TextIOBase],
default_section: str = None, deep: bool = False):
"""Initialize.
: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 config: configures this instance (see class docs)
:param default_section: used as the default section when non given on
the get methds such as :meth:`get_option`
"""
if isinstance(config_file, str):
self.config_file = Path(config_file).expanduser()
else:
self.config_file = config_file
self._parsed_config = PersistedWork('_parsed_config', self)
super().__init__(config=None,
default_section=default_section,
deep=deep)
def _narrow_root(self, conf: Dict[str, Any]) -> Dict[str, str]:
if not isinstance(conf, dict):
raise ConfigurableError(
f'Expecting a root level dict: {self.config_file}')
return conf
@persisted('_parsed_config')
def _get_config(self) -> Dict[str, Dict[str, Any]]:
if hasattr(self, '_ext_config'):
return self._ext_config
if logger.isEnabledFor(logging.INFO):
logger.info(f'loading config: {self.config_file}')
if isinstance(self.config_file, TextIOBase):
conf = json.load(self.config_file)
self.config_file.seek(0)
else:
if not self.config_file.is_file():
raise ConfigurableFileNotFoundError(self.config_file)
with open(self.config_file) as f:
conf = json.load(f)
conf = self._narrow_root(conf)
if logger.isEnabledFor(logging.DEBUG):
logger.debug(f'raw json: {conf}')
has_terminals = True
for k, v in conf.items():
if isinstance(v, dict):
has_terminals = False
break
if has_terminals:
conf = {self.default_section: conf}
return conf
def _set_config(self, source: Dict[str, Any]):
self._ext_config = source
self._parsed_config.clear()
self.invalidate()