Add basic GenericApplication

service_oriented/application/generic_application.py

@@ -0,0 +1,28 @@
+from typing import Dict, Generic, TypeVar
+
+from service_oriented.application.config import BaseConfig
+from service_oriented.application.entry_point_spec import EntryPointSpec
+
+
+C = TypeVar("C", bound=BaseConfig)
+E = TypeVar("E", bound=EntryPointSpec)
+
+
+class GenericApplication(Generic[C]):
+    def __init__(
+        self,
+        config: C,
+        entry_points: Dict[str, E],
+    ):
+        self.config: C = config
+        self.entry_points: Dict[str, E] = entry_points
+
+    def run(self) -> None:
+        entry_point_name = self.config.entry_point
+        entry_point_spec = self.entry_points.get(entry_point_name)
+        if entry_point_spec is None:
+            raise RuntimeError(
+                f"No mapping found for entry point named '{entry_point_name}'"
+            )
+        entry_point = entry_point_spec.build(config=self.config)
+        entry_point.run()

service_oriented_test/application/generic_application_test.py

@@ -0,0 +1,99 @@
+from typing import Callable
+
+import pytest
+
+from service_oriented.application.config import BaseConfig
+from service_oriented.application.entry_point_spec import EntryPointSpec
+from service_oriented.application.generic_application import GenericApplication
+from service_oriented_test.test_helpers import (
+    TEST_DEPLOYMENT_ENVIRONMENT,
+    TEST_ENTRY_POINT,
+)
+
+
+class Config(
+    BaseConfig,
+    env_nested_delimiter="__",
+    env_prefix="generic_application_test_",
+):
+    pass
+
+
+class CallbackEntryPoint:
+    def __init__(
+        self,
+        config: Config,
+        on_run_callback: Callable[[], None],
+    ) -> None:
+        self.config = config
+        self.on_run_callback = on_run_callback
+
+    def run(self) -> None:
+        self.on_run_callback()
+
+
+class Application(GenericApplication[Config]):
+    pass
+
+
+def test_application_init() -> None:
+    config = Config(
+        deployment_environment=TEST_DEPLOYMENT_ENVIRONMENT,
+        entry_point=TEST_ENTRY_POINT,
+    )
+    entry_points = {
+        "other": EntryPointSpec(CallbackEntryPoint),
+    }
+    application = Application(
+        config=config,
+        entry_points=entry_points,
+    )
+    assert config == application.config
+    assert entry_points == application.entry_points
+
+
+def test_application_run() -> None:
+    on_run_called = False
+
+    def on_run_callback() -> None:
+        nonlocal on_run_called
+        on_run_called = True
+
+    config = Config(
+        deployment_environment=TEST_DEPLOYMENT_ENVIRONMENT,
+        entry_point=TEST_ENTRY_POINT,
+    )
+    entry_points = {
+        TEST_ENTRY_POINT: EntryPointSpec(
+            CallbackEntryPoint,
+            on_run_callback=on_run_callback,
+        ),
+    }
+    application = Application(
+        config=config,
+        entry_points=entry_points,
+    )
+
+    application.run()
+    assert on_run_called
+
+
+def test_application_run_raises_for_unknown_entry_point() -> None:
+    config = Config(
+        deployment_environment=TEST_DEPLOYMENT_ENVIRONMENT,
+        entry_point="other",
+    )
+    entry_points = {
+        TEST_ENTRY_POINT: EntryPointSpec(CallbackEntryPoint),
+    }
+    application = Application(
+        config=config,
+        entry_points=entry_points,
+    )
+
+    with pytest.raises(RuntimeError) as exinfo:
+        application.run()
+
+    assert RuntimeError == exinfo.type
+    exception_message = str(exinfo.value)
+    assert "No mapping found for entry point named 'other'" == exception_message

whitelist.txt

@@ -1,5 +1,6 @@
 dirname
 dotenv
 exinfo
+nonlocal
 param
 pydantic