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