Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(config): add setting_initializer decorator #45

Merged
merged 1 commit into from
Jul 13, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
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