"""Different config formats. Currently supported are:
- JSON
- INI
- YAML
- TOML
Round-trip preservation for preserving comments in config files (depending on
third-party libraries). Internally all config formats get mapped on a
:class:`being.utils.NestedDict` data structure which can be accessed through a
`path-like` syntax.
Example:
>>> c = Config()
... c.store('this/is/it', 1234)
... print(c.data)
{'this': {'is': {'it': 1234}}}
"""
import collections
import io
import json
import os
from typing import Tuple, Any, Optional, TextIO, Dict
import ruamel.yaml
import tomlkit
import configobj
from being.utils import NestedDict
SEP: str = '/'
"""Separator character for name <-> key path conversion."""
ROOT_NAME: str = ''
"""Empty string denoting the root config entry."""
[docs]def split_name(name: str) -> Tuple[str, str]:
"""Split name into (head, tail) where tail is everything after the finals
separator (like splitting a filepath in the directory vs. filename part).
Args:
name: Name to split.
Returns:
head and tail tuple
Example:
>>> split_name('this/is/it')
('this/is', 'it')
"""
if SEP not in name:
return '', name
return name.rsplit('/', maxsplit=1)
class _ConfigImpl(NestedDict):
"""Base class for config implementation / interface.
Holds the dict-like data object from the various third-party libraries.
Contains the default_factory() for intermediate levels. Allows for path like
`name` syntax to access nested dicts with string as keys.
Attributes:
data: Original dict-like config data object.
default_factory: Associated default_factory for creating intermediate
elements
Example:
>>> c = _ConfigImpl()
... c.store('this/is/it', 'Hello, world!')
... c.storedefault('this/is/it', 42)
'Hello, world!'
Todo:
Ditch subclassing from :class:`being.utils.NestedDict`. Composition over
inheritance.
"""
def retrieve(self, name: str = ROOT_NAME) -> Any:
"""Retrieve config value for a given name. This can also be an intermediate entry.
Args:
name (optional): Config path name to retrieve value for. Root path
by default.
Returns:
Config value.
"""
if name == ROOT_NAME:
return self.data
keys = tuple(name.split(SEP))
return self[keys]
def store(self, name: str, value: Any):
"""Store value to config.
Args:
name: Config path name to store value under.
"""
keys = tuple(name.split(SEP))
self[keys] = value
def erase(self, name: str):
"""Erase config value for a given `name`.
Args:
name: Config path name to erase.
"""
keys = tuple(name.split(SEP))
del self[keys]
def storedefault(self, name: str, default: Any = None) -> Any:
"""Fetch config value while providing a default value if entry does not
exist. Similar :class:`dict.setdefault`.
Args:
name: Config path name to store value under.
default: Default value to insert if config entry does not exist.
Returns:
Config value.
"""
keys = tuple(name.split(SEP))
return self.setdefault(keys, default)
def loads(self, string: str):
"""Load data from string to :class:`being.configs._ConfigImpl` instance.
Args:
string: String to parse.
"""
raise NotImplementedError
def load(self, stream: TextIO):
"""Load from stream to :class:`being.configs._ConfigImpl` instance.
Args:
stream: Stream like source to read from.
"""
raise NotImplementedError
def dumps(self) -> str:
"""Dumps config data to string."""
raise NotImplementedError
def dump(self, stream: TextIO) -> str:
"""Dump config data to stream.
Args:
stream: Stream like to write to.
"""
raise NotImplementedError
class _TomlConfig(_ConfigImpl):
"""Config implementation for TOML format."""
def __init__(self, data=None):
if data is None:
data = tomlkit.document() # Differs from default_factory=tomlkit.table
super().__init__(data, default_factory=tomlkit.table)
def loads(self, string):
self.data = tomlkit.loads(string)
def load(self, stream):
self.data = tomlkit.loads(stream.read())
def dumps(self):
return tomlkit.dumps(self.data)
def dump(self, stream):
stream.write(tomlkit.dumps(self.data))
class _YamlConfig(_ConfigImpl):
"""Config implementation for YAML format."""
def __init__(self, data=None):
super().__init__(data, default_factory=ruamel.yaml.CommentedMap)
self.yaml = ruamel.yaml.YAML()
def loads(self, string):
data = self.yaml.load(string)
if data is None:
data = ruamel.yaml.CommentedMap()
self.data = data
def load(self, stream):
data = self.yaml.load(stream)
if data is None:
data = ruamel.yaml.CommentedMap()
self.data = data
def dumps(self):
out = io.StringIO()
self.yaml.dump(self.data, stream=out)
out.seek(0)
return out.read()
def dump(self, stream):
self.yaml.dump(self.data, stream)
class _JsonConfig(_ConfigImpl):
"""Config implementation for JSON format. JSON does not support comments!
Getting or setting comments will result in RuntimeError errors.
"""
def __init__(self, data=None):
super().__init__(data, default_factory=dict)
def loads(self, string):
self.data = json.loads(string)
def load(self, stream):
self.data = json.load(stream)
def dumps(self):
return json.dumps(self.data, indent=4)
def dump(self, stream):
return json.dump(self.data, stream, indent=4)
class _IniConfig(_ConfigImpl):
"""Config implementation for INI format. Also supports round trip
preservation. Comments should be possible as well but not implemented right
now.
"""
def __init__(self, data=None):
super().__init__(data, default_factory=configobj.ConfigObj)
def loads(self, string):
buf = io.StringIO(string)
self.data = configobj.ConfigObj(buf)
def load(self, stream):
self.data = configobj.ConfigObj(stream)
def dumps(self):
buf = io.BytesIO()
self.data.write(buf)
buf.seek(0)
return buf.read().decode()
def dump(self, stream):
return self.data.write(stream)
IMPLEMENTATIONS = {
None: _ConfigImpl,
'toml': _TomlConfig,
'yaml': _YamlConfig,
'json': _JsonConfig,
'ini': _IniConfig,
}
[docs]class Config(_ConfigImpl, collections.abc.MutableMapping):
"""Configuration object. Proxy for _ConfigImpl (depending on config format)."""
def __init__(self, data: Optional[dict] = None, configFormat: Optional[str] = None):
"""
Args:
data (optional): Initial data.
configFormat (optional): Config format (if any).
"""
if configFormat not in IMPLEMENTATIONS:
raise ValueError(f'No config implementation for {configFormat}!')
implType = IMPLEMENTATIONS[configFormat]
self.impl: _ConfigImpl = implType(data)
"""Private config implementation."""
@property
def data(self) -> Dict:
"""Underlying dict-like data container."""
return self.impl.data
def __getitem__(self, key):
return self.impl[key]
def __setitem__(self, key, value):
self.impl[key] = value
def __delitem__(self, key):
del self.impl[key]
def __iter__(self):
return iter(self.impl)
def __len__(self):
return len(self.impl)
[docs] def store(self, name: str, value: Any):
self.impl.store(name, value)
[docs] def retrieve(self, name: str = ROOT_NAME) -> Any:
return self.impl.retrieve(name)
[docs] def erase(self, name: str):
return self.impl.erase(name)
[docs] def storedefault(self, name: str, default: Optional[Any] = None):
return self.impl.storedefault(name, default)
[docs] def loads(self, string: str):
self.impl.loads(string)
[docs] def load(self, stream: TextIO):
self.impl.load(stream)
[docs] def dumps(self):
return self.impl.dumps()
[docs] def dump(self, stream: TextIO):
return self.impl.dump(stream)
[docs]class ConfigFile(Config):
"""Config instance which can be loaded / saved to a config file on disk."""
def __init__(self, filepath: str):
"""
Args:
filepath: Associated config file.
"""
super().__init__(configFormat=guess_config_format(filepath))
self.filepath: str = filepath
"""Associated filepath of config file."""
if os.path.exists(self.filepath):
self.reload()
[docs] def save(self):
"""Save data to config file."""
with open(self.filepath, 'w') as fp:
self.impl.dump(fp)
[docs] def reload(self):
"""Load data from config file."""
with open(self.filepath) as fp:
self.impl.load(fp)
def __str__(self):
return f'{type(self).__name__}({self.filepath!r})'