config_ninja.backend
Define the API for config backends.
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__ = ['FormatT', 'dumps', 'loads', 'Backend'] 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 @abc.abstractmethod 75 def get(self) -> str: 76 """Retrieve the configuration as a raw string.""" 77 78 @classmethod 79 def new( 80 cls: type[Backend], 81 *args: Any, 82 **kwargs: Any, 83 ) -> Backend: 84 """Connect a new instance to the backend.""" 85 return cls(*args, **kwargs) 86 87 @abc.abstractmethod 88 async def poll(self, interval: int = 0) -> AsyncIterator[str]: 89 """Poll the configuration for changes.""" 90 yield '' # pragma: no cover 91 92 93logger.debug('successfully imported %s', __name__)
FormatT =
typing.Literal['json', 'raw', 'toml', 'yaml', 'yml']
The supported serialization formats (not including jinja2 templates)
54def dumps(fmt: FormatT, data: dict[str, Any]) -> str: 55 """Serialize the given `data` object to the given `FormatT`.""" 56 try: 57 dump = DUMPERS[fmt] 58 except KeyError as exc: # pragma: no cover 59 raise ValueError(f"unsupported format: '{fmt}'") from exc 60 61 return dump(data)
Serialize the given data object to the given FormatT.
64def loads(fmt: FormatT, raw: str) -> dict[str, Any]: 65 """Deserialize the given `raw` string for the given `FormatT`.""" 66 try: 67 return LOADERS[fmt](raw) 68 except KeyError as exc: # pragma: no cover 69 raise ValueError(f"unsupported format: '{fmt}'") from exc
Deserialize the given raw string for the given FormatT.
72class Backend(abc.ABC): 73 """Define the API for backend implementations.""" 74 75 @abc.abstractmethod 76 def get(self) -> str: 77 """Retrieve the configuration as a raw string.""" 78 79 @classmethod 80 def new( 81 cls: type[Backend], 82 *args: Any, 83 **kwargs: Any, 84 ) -> Backend: 85 """Connect a new instance to the backend.""" 86 return cls(*args, **kwargs) 87 88 @abc.abstractmethod 89 async def poll(self, interval: int = 0) -> AsyncIterator[str]: 90 """Poll the configuration for changes.""" 91 yield '' # pragma: no cover
Define the API for backend implementations.
@abc.abstractmethod
def
get(self) -> str:
75 @abc.abstractmethod 76 def get(self) -> str: 77 """Retrieve the configuration as a raw string."""
Retrieve the configuration as a raw string.