Implement YAML/JSON/TOML/Python config file support

hyperglass/configuration/collect.py

@@ -0,0 +1,119 @@
+"""Collect configurations from files."""
+
+# Standard Library
+import typing as t
+from pathlib import Path
+
+# Project
+from hyperglass.util import run_coroutine_in_new_thread
+from hyperglass.settings import Settings
+from hyperglass.constants import CONFIG_EXTENSIONS
+from hyperglass.exceptions.private import (
+    ConfigError,
+    ConfigMissing,
+    ConfigLoaderMissing,
+)
+
+LoadedConfig = t.Union[t.Dict[str, t.Any], t.List[t.Any], t.Tuple[t.Any, ...]]
+
+
+def find_path(file_name: str, *, required: bool) -> t.Union[Path, None]:
+    """Find the first matching configuration file."""
+    for extension in CONFIG_EXTENSIONS:
+        path = Settings.app_path / f"{file_name}.{extension}"
+        if path.exists():
+            return path
+
+    if required:
+        raise ConfigMissing(file_name, app_path=Settings.app_path)
+    return None
+
+
+def load_dsl(path: Path, *, empty_allowed: bool) -> LoadedConfig:
+    """Verify and load data from DSL (non-python) config files."""
+    loader = None
+    if path.suffix in (".yaml", ".yml"):
+        try:
+            # Third Party
+            import yaml
+
+            loader = yaml.safe_load
+
+        except ImportError:
+            raise ConfigLoaderMissing(path)
+    elif path.suffix == ".toml":
+        try:
+            # Third Party
+            import toml
+
+            loader = toml.load
+
+        except ImportError:
+            raise ConfigLoaderMissing(path)
+
+    elif path.suffix == ".json":
+        # Standard Library
+        import json
+
+        loader = json.load
+
+    if loader is None:
+        raise ConfigLoaderMissing(path)
+
+    with path.open("r") as f:
+        data = loader(f)
+        if data is None and empty_allowed is False:
+            raise ConfigError(
+                "'{!s}' exists, but it is empty and is required to start hyperglass.".format(path),
+            )
+    return data or {}
+
+
+def load_python(path: Path, *, empty_allowed: bool) -> LoadedConfig:
+    """Import configuration from a python configuration file."""
+    # Standard Library
+    import inspect
+    from importlib.util import module_from_spec, spec_from_file_location
+
+    # Load the file as a module.
+    name, _ = path.name.split(".")
+    spec = spec_from_file_location(name, location=path)
+    module = module_from_spec(spec)
+    spec.loader.exec_module(module)
+    # Get all exports that are named 'main' (any case).
+    exports = tuple(getattr(module, e, None) for e in dir(module) if e.lower() == "main")
+    if len(exports) < 1:
+        # Raise an error if there are no exports named main.
+        raise ConfigError(
+            f"'{path!s} exists', but it is missing a variable or function named 'main'"
+        )
+    # Pick the first export named main.
+    main, *_ = exports
+    data = None
+    if isinstance(main, t.Callable):
+        if inspect.iscoroutinefunction(main):
+            # Resolve an async funcion.
+            data = run_coroutine_in_new_thread(main)
+        else:
+            # Resolve a standard function.
+            data = main()
+    elif isinstance(main, (t.Dict, t.List, t.Tuple)):
+        data = main
+
+    if data is None and empty_allowed is False:
+        raise ConfigError(f"'{path!s} exists', but variable or function 'main' is an invalid type")
+    return data or {}
+
+
+def load_config(name: str, *, required: bool) -> LoadedConfig:
+    """Load a configuration file."""
+    path = find_path(name, required=required)
+    if path.suffix == ".py":
+        return load_python(path, empty_allowed=not required)
+    elif path.suffix.replace(".", "") in CONFIG_EXTENSIONS:
+        return load_dsl(path, empty_allowed=not required)
+    raise ConfigError(
+        "{p} has an unsupported file extension. Must be one of {e}",
+        p=path,
+        e=", ".join(CONFIG_EXTENSIONS),
+    )

hyperglass/configuration/tests/__init__.py

@@ -0,0 +1 @@
+"""hyperglass configuration tests."""

hyperglass/configuration/tests/test_collect.py

@@ -0,0 +1,58 @@
+"""Test configuration file collection."""
+
+# Standard Library
+import tempfile
+from pathlib import Path
+
+# Project
+from hyperglass.settings import Settings
+
+# Local
+from ..collect import load_config
+
+TOML = """
+test = "from toml"
+"""
+
+YAML = """
+test: from yaml
+"""
+
+JSON = """
+{"test": "from json"}
+"""
+
+PY_VARIABLE = """
+MAIN = {'test': 'from python variable'}
+"""
+
+PY_FUNCTION = """
+def main():
+    return {'test': 'from python function'}
+"""
+
+PY_COROUTINE = """
+async def main():
+    return {'test': 'from python coroutine'}
+"""
+
+CASES = (
+    ("test.toml", "from toml", TOML),
+    ("test.yaml", "from yaml", YAML),
+    ("test_py_variable.py", "from python variable", PY_VARIABLE),
+    ("test_py_function.py", "from python function", PY_FUNCTION),
+    ("test_py_coroutine.py", "from python coroutine", PY_COROUTINE),
+)
+
+
+def test_collect(monkeypatch):
+    with tempfile.TemporaryDirectory() as directory_name:
+        directory = Path(directory_name)
+        monkeypatch.setattr(Settings, "app_path", directory)
+        for name, value, data in CASES:
+            path = directory / Path(name)
+            with path.open("w") as p:
+                p.write(data)
+            loaded = load_config(path.stem, required=True)
+            assert loaded.get("test") is not None
+            assert loaded["test"] == value

hyperglass/constants.py

@@ -21,6 +21,8 @@ TARGET_JUNIPER_ASPATH = ("juniper", "juniper_junos")
 
 SUPPORTED_STRUCTURED_OUTPUT = ("juniper", "arista_eos")
 
+CONFIG_EXTENSIONS = ("py", "yaml", "yml", "json", "toml")
+
 STATUS_CODE_MAP = {"warning": 400, "error": 400, "danger": 500}
 
 DNS_OVER_HTTPS = {

hyperglass/exceptions/private.py

@@ -2,6 +2,10 @@
 
 # Standard Library
 from typing import Any, Dict, List
+from pathlib import Path
+
+# Project
+from hyperglass.constants import CONFIG_EXTENSIONS
 
 # Local
 from ._common import ErrorLevel, PrivateHyperglassError
@@ -63,15 +67,28 @@ class ConfigInvalid(PrivateHyperglassError):
 class ConfigMissing(PrivateHyperglassError):
     """Raised when a required config file or item is missing or undefined."""
 
-    def __init__(self, missing_item: Any) -> None:
-        """Show the missing configuration item."""
-        super().__init__(
+    def __init__(self, file_name: str, *, app_path: Path) -> None:
+        """Customize error message."""
+        message = " ".join(
             (
-                "{item} is missing or undefined and is required to start hyperglass. "
-                "Please consult the installation documentation."
-            ),
-            item=missing_item,
+                file_name.capitalize(),
+                "file is missing in",
+                f"'{app_path!s}', and is required to start hyperglass.",
+                "Supported file names are:",
+                ", ".join(f"'{file_name}.{e}'" for e in CONFIG_EXTENSIONS),
+                ". Please consult the installation documentation.",
+            )
         )
+        super().__init__(message)
+
+
+class ConfigLoaderMissing(PrivateHyperglassError):
+    """Raised when a configuration file is using a file extension that requires a missing loader."""
+
+    def __init__(self, path: Path, /) -> None:
+        """Customize error message."""
+        message = "'{path}' requires a {loader} loader, but it is not installed"
+        super().__init__(message=message, path=path, loader=path.suffix.strip("."))
 
 
 class ConfigError(PrivateHyperglassError):

hyperglass/models/system.py

@@ -29,6 +29,8 @@ class HyperglassSettings(BaseSettings):
 
         env_prefix = "hyperglass_"
 
+    config_file_names: t.ClassVar[t.Tuple[str, ...]] = ("config", "devices", "directives")
+
     debug: bool = False
     dev_mode: bool = False
     app_path: DirectoryPath

hyperglass/util/__init__.py

@@ -362,3 +362,24 @@ def compare_init(obj_a: object, obj_b: object) -> bool:
         obj_b.__init__.__annotations__.pop("self", None)
         return compare_dicts(obj_a.__init__.__annotations__, obj_b.__init__.__annotations__)
     return False
+
+
+def run_coroutine_in_new_thread(coroutine: t.Coroutine) -> t.Any:
+    """Run an async function in a separate thread and get the result."""
+    # Standard Library
+    import asyncio
+    import threading
+
+    class Resolver(threading.Thread):
+        def __init__(self, coro: t.Coroutine) -> None:
+            self.result: t.Any = None
+            self.coro: t.Coroutine = coro
+            super().__init__()
+
+        def run(self):
+            self.result = asyncio.run(self.coro())
+
+    thread = Resolver(coroutine)
+    thread.start()
+    thread.join()
+    return thread.result

hyperglass/util/tests/test_utilities.py

@@ -1,7 +1,9 @@
 """Test generic utilities."""
+# Standard Library
+import asyncio
 
 # Local
-from .. import compare_init, compare_dicts
+from .. import compare_init, compare_dicts, run_coroutine_in_new_thread
 
 
 def test_compare_dicts():
@@ -53,3 +55,15 @@ def test_compare_init():
     )
     for a, b, expected in checks:
         assert compare_init(a, b) is expected
+
+
+def test_run_coroutine_in_new_thread():
+    async def sleeper():
+        await asyncio.sleep(5)
+
+    async def test():
+        return True
+
+    asyncio.run(sleeper())
+    result = run_coroutine_in_new_thread(test)
+    assert result is True