config_ninja.backend

  1"""Define the API for config backends."""
  2
  3from __future__ import annotations
  4
  5import abc
  6import json
  7import logging
  8import typing
  9from typing import Any, AsyncIterator, Callable, Dict
 10
 11import tomlkit as toml
 12import yaml
 13
 14__all__ = ['Backend', 'FormatT', 'dumps', 'loads']
 15
 16logger = logging.getLogger(__name__)
 17
 18FormatT = typing.Literal['json', 'raw', 'toml', 'yaml', 'yml']
 19"""The supported serialization formats (not including `jinja2` templates)"""
 20
 21# note: `3.8` was not respecting `from __future__ import annotations` for delayed evaluation
 22LoadT = Callable[[str], Dict[str, Any]]
 23DumpT = Callable[[Dict[str, Any]], str]
 24
 25
 26def load_raw(raw: str) -> dict[str, str]:
 27    """Treat the string as raw text."""
 28    return {'content': raw}
 29
 30
 31def dump_raw(data: dict[str, str]) -> str:
 32    """Get the `'content'` key from the given `dict`."""
 33    return data['content']
 34
 35
 36LOADERS: dict[FormatT, LoadT] = {
 37    'json': json.loads,
 38    'raw': load_raw,
 39    'toml': toml.loads,
 40    'yaml': yaml.safe_load,
 41    'yml': yaml.safe_load,
 42}
 43
 44DUMPERS: dict[FormatT, DumpT] = {
 45    'json': json.dumps,
 46    'raw': dump_raw,
 47    'toml': toml.dumps,  # pyright: ignore[reportUnknownMemberType]
 48    'yaml': yaml.dump,
 49    'yml': yaml.dump,
 50}
 51
 52
 53def dumps(fmt: FormatT, data: dict[str, Any]) -> str:
 54    """Serialize the given `data` object to the given `FormatT`."""
 55    try:
 56        dump = DUMPERS[fmt]
 57    except KeyError as exc:  # pragma: no cover
 58        raise ValueError(f"unsupported format: '{fmt}'") from exc
 59
 60    return dump(data)
 61
 62
 63def loads(fmt: FormatT, raw: str) -> dict[str, Any]:
 64    """Deserialize the given `raw` string for the given `FormatT`."""
 65    try:
 66        return LOADERS[fmt](raw)
 67    except KeyError as exc:  # pragma: no cover
 68        raise ValueError(f"unsupported format: '{fmt}'") from exc
 69
 70
 71class Backend(abc.ABC):
 72    """Define the API for backend implementations."""
 73
 74    def __repr__(self) -> str:
 75        """Represent the backend object as its invocation.
 76
 77        >>> example = ExampleBackend('an example')
 78        >>> example
 79        ExampleBackend(source='an example')
 80        """
 81        annotations = (klass := self.__class__).__annotations__
 82        annotations.pop('return', None)
 83
 84        args = ', '.join(f'{key}={getattr(self, key)!r}' for key in annotations if hasattr(self, key))
 85        return f'{klass.__name__}({args})'
 86
 87    @abc.abstractmethod
 88    def __str__(self) -> str:
 89        """When formatted as a string, represent the backend as the identifier of its source."""
 90
 91    @abc.abstractmethod
 92    def get(self) -> str:
 93        """Retrieve the configuration as a raw string."""
 94
 95    @classmethod
 96    def new(
 97        cls: type[Backend],
 98        *args: Any,
 99        **kwargs: Any,
100    ) -> Backend:
101        """Connect a new instance to the backend."""
102        return cls(*args, **kwargs)
103
104    @abc.abstractmethod
105    async def poll(self, interval: int = 0) -> AsyncIterator[str]:
106        """Poll the configuration for changes."""
107        yield ''  # pragma: no cover
108
109
110logger.debug('successfully imported %s', __name__)