Skip to content

Commit

Permalink
feat(config): add setting_initializer decorator (#45)
Browse files Browse the repository at this point in the history
Add the `sghi.config.setting_initializer` decorator function. This
should simplify the construction of simple setting initializers.
  • Loading branch information
kennedykori authored Jul 13, 2024
1 parent de349b8 commit 9dd4a49
Show file tree
Hide file tree
Showing 2 changed files with 199 additions and 8 deletions.
158 changes: 151 additions & 7 deletions src/sghi/config/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,12 +5,21 @@
import logging
from abc import ABCMeta, abstractmethod
from collections.abc import Callable
from functools import update_wrapper, wraps
from logging import Logger
from typing import TYPE_CHECKING, Any, Final, final

from typing_extensions import override

from ..exceptions import SGHIError
from ..task import Task, pipe
from ..utils import ensure_not_none, ensure_not_none_nor_empty, type_fqn
from ..utils import (
ensure_callable,
ensure_instance_of,
ensure_not_none,
ensure_not_none_nor_empty,
type_fqn,
)

if TYPE_CHECKING:
from collections.abc import Mapping, Sequence
Expand Down Expand Up @@ -74,6 +83,80 @@ def register(f: _Initializer_Factory) -> _Initializer_Factory:
return f


def setting_initializer(
*,
setting: str,
has_secrets: bool = False,
) -> Callable[[Callable[[Any], Any]], _Initializer_Factory]:
"""Mark/Decorate a callable as a :class:`~sghi.config.SettingInitializer`.
This decorator converts a callable into a factory function that supplies
``SettingInitializer`` instances with the same behaviour/semantics as the
decorated function. The decorator is designed to be used together with the
:meth:`@register<sghi.config.register>` decorator.
.. important::
The decorated callable *MUST* accept at least one argument but have
at *MOST* one required argument.
.. note::
This decorator returns a factory function and NOT a
``SettingInitializer``. To get a ``SettingInitializer`` instance,
invoke the decorated function without providing any arguments. See the
usage example below.
Here's a usage example:
.. code-block:: python
:linenos:
from sghi.config import *
@register # This is optional
@setting_initializer(setting="USERNAME", has_secrets=True)
def username_initializer(username: str | None) -> str:
if not username:
err_msg: str = "'USERNAME' MUST be provided."
raise SettingRequiredError(setting="USERNAME", message=err_msg)
return username
# Get a `SettingInitializer` instance
initializer = username_initializer()
assert isinstance(initializer, SettingInitializer)
initializer("C-3PO") # OK
initializer(None) # Error: SettingRequiredError
:param setting: The setting to be initialized using the resulting
``SettingInitializer``. This MUST be a non-empty string.
:param has_secrets: A flag indicating whether the setting to be initialized
contains sensitive information. This MUST be a boolean value. Defaults
to ``False``.
:return: A factory function that supplies ``SettingInitializer`` instances
with the same behaviour/semantics as the decorated callable.
"""

def wrap(f: Callable[[Any], Any]) -> _Initializer_Factory:
@wraps(
f,
assigned=("__module__", "__name__", "__qualname__", "__doc__"),
)
def setting_initializer_factory() -> SettingInitializer:
return _SettingInitializeOfCallable(
source_callable=f,
setting=setting,
has_secrets=has_secrets,
)

return setting_initializer_factory

return wrap


# =============================================================================
# EXCEPTIONS
# =============================================================================
Expand Down Expand Up @@ -228,6 +311,57 @@ def setting(self) -> str:
"""


# =============================================================================
# SETTING INITIALIZER IMPLEMENTATIONS
# =============================================================================


@final
class _SettingInitializeOfCallable(SettingInitializer):
__slots__ = ("_source_callable", "_setting", "_has_secretes", "__dict__")

def __init__(
self,
source_callable: Callable[[Any], Any],
setting: str,
has_secrets: bool = False,
):
super().__init__()
self._source_callable: Callable[[Any], Any]
self._source_callable = ensure_callable(
source_callable,
message="'source_callable' MUST be a callable object.",
)
self._setting: str = ensure_not_none_nor_empty(
value=ensure_instance_of(
value=setting,
klass=str,
message="'setting' MUST be a string.",
),
message="'setting' MUST NOT be an empty string.",
)
self._has_secrets: bool = ensure_instance_of(
value=has_secrets,
klass=bool,
message="'has_secrets' MUST be a boolean value.",
)
update_wrapper(self, self._source_callable)

@property
@override
def has_secrets(self) -> bool:
return self._has_secrets

@property
@override
def setting(self) -> str:
return self._setting

@override
def execute(self, an_input: Any) -> Any:
return self._source_callable(an_input)


# =============================================================================
# CONFIG INTERFACE
# =============================================================================
Expand Down Expand Up @@ -468,15 +602,18 @@ def __init__(self, source_config: Config) -> None:
ensure_not_none(source_config, "'source_config' MUST not be None.")
self._source_config: Config = source_config

@override
def __contains__(self, __setting: str, /) -> bool:
"""Check for the availability of a setting."""
return self._source_config.__contains__(__setting)

def __getattr__(self, __setting: str, /) -> Any: # noqa: ANN401
@override
def __getattr__(self, __setting: str, /) -> Any:
"""Make settings available using the dot operator."""
return self._source_config.__getattr__(__setting)

def get(self, setting: str, default: Any = None) -> Any: # noqa: ANN401
@override
def get(self, setting: str, default: Any = None) -> Any:
return self._source_config.get(setting=setting, default=default)

def set_source(self, source_config: Config) -> None:
Expand Down Expand Up @@ -533,18 +670,21 @@ def __init__(
self._logger: Logger = logging.getLogger(type_fqn(self.__class__))
self._run_initializers()

@override
def __contains__(self, __setting: str, /) -> bool:
"""Check for the availability of a setting."""
return self._settings.__contains__(__setting)

def __getattr__(self, __setting: str, /) -> Any: # noqa: ANN401
@override
def __getattr__(self, __setting: str, /) -> Any:
"""Make settings available using the dot operator."""
try:
return self._settings[__setting]
except KeyError:
raise NoSuchSettingError(setting=__setting) from None

def get(self, setting: str, default: Any = None) -> Any: # noqa: ANN401
@override
def get(self, setting: str, default: Any = None) -> Any:
"""Retrieve the value of the given setting or return the given default
if no such setting exists in this ``Config`` instance.
Expand Down Expand Up @@ -620,6 +760,7 @@ def __init__(self, err_msg: str | None = None):
"""
self._err_msg: str | None = err_msg

@override
def __contains__(self, __setting: str, /) -> bool:
"""Raise a ``NotSetupError`` when trying to check for the availability
of a setting.
Expand All @@ -631,7 +772,8 @@ def __contains__(self, __setting: str, /) -> bool:
"""
return self._raise(err_msg=self._err_msg)

def __getattr__(self, __setting: str, /) -> Any: # noqa: ANN401
@override
def __getattr__(self, __setting: str, /) -> Any:
"""Raise a ``NotSetupError`` when trying to access any setting.
:param __setting: The name of the setting value to retrieve.
Expand All @@ -642,7 +784,8 @@ def __getattr__(self, __setting: str, /) -> Any: # noqa: ANN401
"""
return self._raise(err_msg=self._err_msg)

def get(self, setting: str, default: Any = None) -> Any: # noqa: ANN401
@override
def get(self, setting: str, default: Any = None) -> Any:
"""Raise a ``NotSetupError`` when trying to access any setting.
:param setting: The name of the setting value to retrieve.
Expand Down Expand Up @@ -686,4 +829,5 @@ def _raise(err_msg: str | None) -> Never:
"SettingRequiredError",
"get_registered_initializer_factories",
"register",
"setting_initializer",
]
49 changes: 48 additions & 1 deletion test/sghi/config_tests.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,10 @@
from __future__ import annotations

from typing import TYPE_CHECKING, Any
from unittest import TestCase

import pytest
from typing_extensions import override

from sghi.config import (
Config,
Expand All @@ -13,6 +16,7 @@
SettingRequiredError,
get_registered_initializer_factories,
register,
setting_initializer,
)
from sghi.utils import ensure_predicate

Expand All @@ -24,14 +28,22 @@
# =============================================================================


@register
@setting_initializer(setting="DB_USERNAME")
def db_username_initializer(username: str | None) -> str:
return username or "postgres"


@register
class DBPortInitializer(SettingInitializer):
__slots__ = ()

@property
@override
def setting(self) -> str:
return "DB_PORT"

@override
def execute(self, an_input: int | str | None) -> int:
match an_input:
case None:
Expand Down Expand Up @@ -61,13 +73,16 @@ class DBPasswordInitializer(SettingInitializer):
__slots__ = ()

@property
@override
def has_secrets(self) -> bool:
return True

@property
@override
def setting(self) -> str:
return "DB_PASSWORD"

@override
def execute(self, an_input: str | None) -> str:
if not an_input:
_err_msg: str = f"'{self.setting}' is required."
Expand All @@ -85,11 +100,43 @@ def test_get_registered_initializer_factories_return_value() -> None:
""":func:`get_registered_initializer_factories` should return all
initializer factories decorated using the :func:`register` decorator.
"""
assert len(get_registered_initializer_factories()) == 1
assert len(get_registered_initializer_factories()) == 2
for init_factory in get_registered_initializer_factories():
assert isinstance(init_factory(), SettingInitializer)


def test_setting_initializer_return_value() -> None:
""":func:`setting_initializer` should return a factory function that
supplies ``SettingInitializer`` instances with the same properties as those
supplied on the decorator and the same semantics as the decorated function.
"""

@setting_initializer(setting="USERNAME", has_secrets=True)
def username_initializer(username: str | None) -> str:
if not username:
_err_msg: str = "'USERNAME' MUST NOT be a None nor empty string."
raise SettingRequiredError(setting="USERNAME", message=_err_msg)
return username

initializer1: SettingInitializer = username_initializer()
initializer2: SettingInitializer = db_username_initializer()

assert isinstance(initializer1, SettingInitializer)
assert isinstance(initializer2, SettingInitializer)
assert initializer1.has_secrets
assert not initializer2.has_secrets
assert initializer1.setting == "USERNAME"
assert initializer2.setting == "DB_USERNAME"

assert initializer1("C-3PO") == "C-3PO"
with pytest.raises(SettingRequiredError) as exc_info:
initializer1(None)
assert exc_info.value.setting == "USERNAME"

assert initializer2(None) == "postgres"
assert initializer2("app-db-admin") == "app-db-admin"


class TestConfig(TestCase):
"""
Tests of the :class:`Config` interface default method implementations.
Expand Down

0 comments on commit 9dd4a49

Please sign in to comment.