Add EntryPointSpec

service_oriented/application/entry_point_spec.py

@@ -0,0 +1,44 @@
+import sys
+from typing import Protocol, Type, TypeVar
+
+
+if sys.version_info >= (3, 10):  # pragma: no cover
+    from typing import ParamSpec
+else:  # pragma: no cover
+    from typing_extensions import ParamSpec
+
+from service_oriented.application.config import BaseConfig
+
+
+P = ParamSpec("P")
+C = TypeVar("C", bound=BaseConfig)
+
+
+class _EntryPointClass(Protocol[P]):
+    def __init__(self, config: C, *args: P.args, **kwargs: P.kwargs) -> None:
+        ...  # pragma: no cover
+
+    def run(self) -> None:
+        ...  # pragma: no cover
+
+
+class EntryPointSpec:
+    def __init__(
+        self,
+        cls: Type[_EntryPointClass[P]],
+        *args: P.args,
+        **kwargs: P.kwargs,
+    ) -> None:
+        self.args = args
+        self.cls = cls
+        self.kwargs = kwargs
+
+    def build(self, config: C) -> _EntryPointClass:
+        return self.cls(config, *self.args, **self.kwargs)
+
+    def __repr__(self) -> str:
+        class_name = self.__class__.__name__
+        args_strings = [f"{value!r}" for value in self.args]
+        option_strings = [f"{key}={value!r}" for key, value in self.kwargs.items()]
+        args_repr = ", ".join([self.cls.__name__] + args_strings + option_strings)
+        return f"{class_name}({args_repr})"

service_oriented_test/application/entry_point_spec_test.py

@@ -0,0 +1,56 @@
+from typing import cast
+
+from service_oriented.application.config import BaseConfig
+from service_oriented.application.entry_point_spec import EntryPointSpec
+from service_oriented_test.test_helpers import (
+    TEST_DEPLOYMENT_ENVIRONMENT,
+    TEST_ENTRY_POINT,
+)
+
+
+class Config(BaseConfig, env_nested_delimiter="__", env_prefix="entry_point_spec_test"):
+    pass
+
+
+class DummyEntryPoint:
+    def __init__(self, config: Config, arg_1: str, arg_2: str):
+        self.config = config
+        self.arg_1 = arg_1
+        self.arg_2 = arg_2
+
+    def run(self) -> None:
+        pass
+
+
+def test_init() -> None:
+    entry_point_class = DummyEntryPoint
+    arg_1 = "arg_1"
+    arg_2 = "arg_2"
+    entry_point_spec = EntryPointSpec(entry_point_class, arg_1, arg_2=arg_2)
+    assert entry_point_class == entry_point_spec.cls
+    assert (arg_1,) == entry_point_spec.args
+    assert {"arg_2": arg_2} == entry_point_spec.kwargs
+
+
+def test_build() -> None:
+    entry_point_class = DummyEntryPoint
+    arg_1 = "arg_1"
+    arg_2 = "arg_2"
+    entry_point_spec = EntryPointSpec(entry_point_class, arg_1, arg_2=arg_2)
+    config = Config(
+        deployment_environment=TEST_DEPLOYMENT_ENVIRONMENT,
+        entry_point=TEST_ENTRY_POINT,
+    )
+    entry_point = cast(DummyEntryPoint, entry_point_spec.build(config))
+    assert config == entry_point.config
+    assert arg_1 == entry_point.arg_1
+    assert arg_2 == entry_point.arg_2
+
+
+def test_repr() -> None:
+    entry_point_class = DummyEntryPoint
+    arg_1 = "arg_1"
+    arg_2 = "arg_2"
+    entry_point_spec = EntryPointSpec(entry_point_class, arg_1, arg_2=arg_2)
+    expected_repr = "EntryPointSpec(DummyEntryPoint, 'arg_1', arg_2='arg_2')"
+    assert expected_repr == repr(entry_point_spec)

whitelist.txt

@@ -1,4 +1,5 @@
 dirname
 dotenv
 exinfo
+param
 pydantic