config_ninja.controller

  1"""Define a controller class for operating on a `config_ninja.settings.ObjectSpec`."""
  2
  3from __future__ import annotations
  4
  5import logging
  6import typing
  7
  8import jinja2
  9
 10from config_ninja import settings, systemd
 11from config_ninja.backend import FormatT, dumps, loads
 12from config_ninja.settings import ObjectSpec
 13
 14try:
 15    from typing import TypeAlias  # type: ignore[attr-defined,unused-ignore]
 16except ImportError:  # pragma: no cover
 17    from typing_extensions import TypeAlias  # type: ignore[assignment,attr-defined,unused-ignore]
 18
 19__all__ = ['BackendController', 'ErrorHandler']
 20
 21logger = logging.getLogger(__name__)
 22
 23ActionType: TypeAlias = typing.Callable[[str], typing.Any]
 24ErrorHandler: TypeAlias = typing.Callable[[typing.Dict[typing.Any, typing.Any]], typing.ContextManager[None]]
 25
 26
 27class BackendController:
 28    """Define logic for operating a configuration `ObjectSpec`."""
 29
 30    key: str
 31    """The key of the configuration object in the settings file."""
 32
 33    logger: logging.Logger
 34    """Each `BackendController` has its own logger (named `"config_ninja.controller:{key}"`)."""
 35
 36    spec: ObjectSpec
 37    """Full specification for the configuration object."""
 38
 39    def __init__(self, spec: ObjectSpec, key: str) -> None:
 40        """Ensure the parent directory of the destination path exists; initialize a logger."""
 41        self.logger = logging.getLogger(f'{__name__}:{key}')
 42        self.key = key
 43        self.spec = spec
 44
 45        spec.dest.path.parent.mkdir(parents=True, exist_ok=True)
 46
 47    def __str__(self) -> str:
 48        """Represent the controller as its backend populating its destination."""
 49        return f'{self.spec.source.backend} ({self.spec.source.format}) -> {self.spec.dest}'
 50
 51    def _do(self, action: ActionType, data: dict[str, typing.Any]) -> None:
 52        if self.spec.dest.is_template:
 53            assert isinstance(self.spec.dest.format, jinja2.Template)  # noqa: S101  # 👈 for static analysis
 54            action(self.spec.dest.format.render(data))
 55        else:
 56            fmt: FormatT = self.spec.dest.format  # type: ignore[assignment]
 57            action(dumps(fmt, data))
 58
 59    @classmethod
 60    def from_settings(
 61        cls, conf_settings: settings.Config, key: str, handle_key_errors: ErrorHandler
 62    ) -> BackendController:
 63        """Create a `BackendController` instance from the given settings."""
 64        cfg_obj = conf_settings.settings.OBJECTS[key]
 65        with handle_key_errors(cfg_obj):  # type: ignore[arg-type]
 66            spec = ObjectSpec.from_primitives(cfg_obj, conf_settings.engine)
 67        return cls(spec, key)
 68
 69    def get(self, do_print: typing.Callable[[str], typing.Any]) -> None:
 70        """Retrieve and print the value of the configuration object.
 71
 72        Any `config_ninja.settings.poe.Hook`s will be skipped; they are only executed by the `BackendController.write()`
 73        and `BackendController.awrite()` methods.
 74        """
 75        data = loads(self.spec.source.format, self.spec.source.backend.get())
 76        self._do(do_print, data)
 77        for hook in self.spec.hooks:
 78            self.logger.debug('would execute: %s', hook)
 79
 80    async def aget(self, do_print: typing.Callable[[str], typing.Any]) -> None:
 81        """Poll to retrieve the latest configuration object, and print on each update.
 82
 83        Any `config_ninja.settings.poe.Hook`s will be skipped; they are only executed by the `BackendController.write()`
 84        and `BackendController.awrite()` methods.
 85        """
 86        if systemd.AVAILABLE:  # pragma: no cover
 87            systemd.notify()
 88
 89        async for content in self.spec.source.backend.poll():
 90            data = loads(self.spec.source.format, content)
 91            self._do(do_print, data)
 92            for hook in self.spec.hooks:
 93                self.logger.debug('would execute: %s', hook)
 94
 95    def write(self) -> None:
 96        """Retrieve the latest value of the configuration object, and write to file.
 97
 98        If the `config_ninja.settings.ObjectSpec` provides any `config_ninja.settings.poe.Hook`s, they will be executed
 99        after writing the configuration object to the destination file.
100        """
101        data = loads(self.spec.source.format, self.spec.source.backend.get())
102        self._do(self.spec.dest.path.write_text, data)
103        for hook in self.spec.hooks:
104            hook()
105
106    async def awrite(self) -> None:
107        """Poll to retrieve the latest configuration object, and write to file on each update.
108
109        If the `config_ninja.settings.ObjectSpec` provides any `config_ninja.settings.poe.Hook`s, they will be executed
110        after writing the configuration object to the destination file.
111        """
112        if systemd.AVAILABLE:  # pragma: no cover
113            systemd.notify()
114
115        async for content in self.spec.source.backend.poll():
116            data = loads(self.spec.source.format, content)
117            self._do(self.spec.dest.path.write_text, data)
118            for hook in self.spec.hooks:
119                hook()
120
121
122logger.debug('successfully imported %s', __name__)