powerful configuration management

kvmd/yamlconf/__init__.py

@@ -0,0 +1,105 @@
+import json
+
+from typing import Tuple
+from typing import List
+from typing import Dict
+from typing import Callable
+from typing import Optional
+from typing import Any
+
+
+# =====
+def build_raw_from_options(options: List[str]) -> Dict[str, Any]:
+    raw: Dict[str, Any] = {}
+    for option in options:
+        (key, value) = (option.split("=", 1) + [None])[:2]  # type: ignore
+        if len(key.strip()) == 0:
+            raise ValueError("Empty option key (required 'key=value' instead of '{}')".format(option))
+        if value is None:
+            raise ValueError("No value for key '{}'".format(key))
+
+        section = raw
+        subs = list(map(str.strip, key.split("/")))
+        for sub in subs[:-1]:
+            section.setdefault(sub, {})
+            section = section[sub]
+        section[subs[-1]] = _parse_value(value)
+    return raw
+
+
+def _parse_value(value: str) -> Any:
+    value = value.strip()
+    if (
+        not value.isdigit()
+        and value not in ["true", "false", "null"]
+        and not value.startswith(("{", "[", "\""))
+    ):
+        value = "\"{}\"".format(value)
+    return json.loads(value)
+
+
+# =====
+class Section(dict):
+    def __init__(self) -> None:
+        dict.__init__(self)
+        self.__meta: Dict[str, Dict[str, Any]] = {}
+
+    def _set_meta(self, name: str, default: Any, help: str) -> None:  # pylint: disable=redefined-builtin
+        self.__meta[name] = {
+            "default": default,
+            "help":    help,
+        }
+
+    def _get_default(self, name: str) -> Any:
+        return self.__meta[name]["default"]
+
+    def _get_help(self, name: str) -> str:
+        return self.__meta[name]["help"]
+
+    def __getattribute__(self, name: str) -> Any:
+        if name in self:
+            return self[name]
+        else:  # For pickling
+            return dict.__getattribute__(self, name)
+
+
+class Option:
+    __type = type
+
+    def __init__(self, default: Any, help: str="", type: Optional[Callable[[Any], Any]]=None) -> None:  # pylint: disable=redefined-builtin
+        self.default = default
+        self.help = help
+        self.type: Callable[[Any], Any] = (type or (self.__type(default) if default is not None else str))  # type: ignore
+
+    def __repr__(self) -> str:
+        return "<Option(default={self.default}, type={self.type}, help={self.help})>".format(self=self)
+
+
+# =====
+def make_config(raw: Dict[str, Any], scheme: Dict[str, Any], _keys: Tuple[str, ...]=()) -> Section:
+    if not isinstance(raw, dict):
+        raise ValueError("The node '{}' must be a dictionary".format("/".join(_keys) or "/"))
+
+    config = Section()
+    for (key, option) in scheme.items():
+        full_key = _keys + (key,)
+        full_name = "/".join(full_key)
+
+        if isinstance(option, Option):
+            value = raw.get(key, option.default)
+            try:
+                value = option.type(value)
+            except Exception:
+                raise ValueError("Invalid value '{value}' for key '{key}'".format(key=full_name, value=value))
+            config[key] = value
+            config._set_meta(  # pylint: disable=protected-access
+                name=key,
+                default=option.default,
+                help=option.help,
+            )
+        elif isinstance(option, dict):
+            config[key] = make_config(raw.get(key, {}), option, full_key)
+        else:
+            raise RuntimeError("Incorrect scheme definition for key '{}':"
+                               " the value is {}, not dict or Option()".format(full_name, type(option)))
+    return config

kvmd/yamlconf/dumper.py

@@ -0,0 +1,41 @@
+# pylint: skip-file
+# infinite recursion
+
+
+import operator
+
+from typing import Tuple
+from typing import List
+from typing import Any
+
+import yaml
+
+from . import Section
+
+
+# =====
+def make_config_dump(config: Section) -> str:
+    return "\n".join(_inner_make_dump(config))
+
+
+def _inner_make_dump(config: Section, _path: Tuple[str, ...]=()) -> List[str]:
+    lines = []
+    for (key, value) in sorted(config.items(), key=operator.itemgetter(0)):
+        indent = len(_path) * "    "
+        if isinstance(value, Section):
+            lines.append("{}{}:".format(indent, key))
+            lines += _inner_make_dump(value, _path + (key,))
+            lines.append("")
+        else:
+            default = config._get_default(key)  # pylint: disable=protected-access
+            comment = config._get_help(key)  # pylint: disable=protected-access
+            if default == value:
+                lines.append("{}{}: {} # {}".format(indent, key, _make_yaml(value), comment))
+            else:
+                lines.append("{}# {}: {} # {}".format(indent, key, _make_yaml(default), comment))
+                lines.append("{}{}: {}".format(indent, key, _make_yaml(value)))
+    return lines
+
+
+def _make_yaml(value: Any) -> str:
+    return yaml.dump(value, allow_unicode=True).replace("\n...\n", "").strip()

kvmd/yamlconf/loader.py

@@ -0,0 +1,31 @@
+import os
+
+from typing import IO
+from typing import Any
+
+import yaml
+import yaml.loader
+import yaml.nodes
+
+
+# =====
+def load_yaml_file(path: str) -> Any:
+    with open(path) as yaml_file:
+        try:
+            return yaml.load(yaml_file, _YamlLoader)
+        except Exception:
+            # Reraise internal exception as standard ValueError and show the incorrect file
+            raise ValueError("Incorrect YAML syntax in file '{}'".format(path))
+
+
+class _YamlLoader(yaml.loader.Loader):  # pylint: disable=too-many-ancestors
+    def __init__(self, yaml_file: IO) -> None:
+        yaml.loader.Loader.__init__(self, yaml_file)
+        self.__root = os.path.dirname(yaml_file.name)
+
+    def include(self, node: yaml.nodes.Node) -> str:
+        path = os.path.join(self.__root, self.construct_scalar(node))  # pylint: disable=no-member
+        return load_yaml_file(path)
+
+
+_YamlLoader.add_constructor("!include", _YamlLoader.include)  # pylint: disable=no-member