Use svcs as service registry

docs/api.md

@@ -7,5 +7,5 @@
         members:
         - Wheke
         - Pod
-        - Service
-        - ServiceRegistry
+        - aget_service
+        - get_service

docs/index.md

@@ -27,6 +27,7 @@
 **Wheke** is an opinionated framework to build small scale self-hosted applications and it stands on top of well known packages:
 
 - [FastAPI](https://fastapi.tiangolo.com/) to build the web services.
+- [svcs](https://svcs.hynek.me/) to have a robust service registry.
 - [Typer](https://typer.tiangolo.com/) to build great cli.
 - [Pydantic](https://docs.pydantic.dev/latest/) to build schemas and `.env` based settings.
 - [Rich](https://rich.readthedocs.io/en/stable/) to make your app shine.

pyproject.toml

@@ -27,6 +27,7 @@ dependencies = [
   "pydantic",
   "pydantic-settings",
   "rich",
+  "svcs",
   "typer",
 ]
 

src/wheke/__about__.py

@@ -1 +1 @@
-__version__ = "0.0.1a9"
+__version__ = "0.0.1a10"

src/wheke/__init__.py

@@ -1,10 +1,10 @@
 from wheke.core import Wheke
 from wheke.pod import Pod
-from wheke.service import Service, ServiceRegistry
+from wheke.service import aget_service, get_service
 
 __all__ = [
     "Pod",
     "Wheke",
-    "Service",
-    "ServiceRegistry",
+    "aget_service",
+    "get_service",
 ]

src/wheke/core.py

@@ -6,7 +6,7 @@
 
 from wheke.cli import empty_callback, version
 from wheke.pod import Pod
-from wheke.service import ServiceRegistry
+from wheke.service import get_service_registry
 from wheke.settings import settings
 
 
@@ -37,7 +37,7 @@ def add_pod(self, pod_to_add: Pod | str) -> None:
             pod = pod_to_add
 
         for service_type, service_factory in pod.services:
-            ServiceRegistry.register(service_type, service_factory)
+            get_service_registry().register_factory(service_type, service_factory)
 
         self.pods.append(pod)
 

src/wheke/pod.py

@@ -4,9 +4,7 @@
 from fastapi import APIRouter
 from typer import Typer
 
-from wheke.service import Service
-
-ServiceList = list[tuple[type[Service], Callable]]
+ServiceList = list[tuple[type, Callable]]
 
 
 class Pod:

src/wheke/service.py

@@ -1,46 +1,19 @@
-from abc import ABC
-from collections.abc import Callable
-from typing import ClassVar
+from typing import TypeVar
 
-from wheke.exceptions import ServiceTypeNotRegisteredError
+from svcs import Container, Registry
 
+T = TypeVar("T")
 
-class Service(ABC):
-    """
-    The Service base class that all Wheke
-    services must implement.
-    """
+_registry = Registry()
 
-    pass
 
+def get_service_registry() -> Registry:
+    return _registry
 
-class ServiceRegistry:
-    """
-    ServiceRegistry used to register and get services.
-    """
 
-    _registry: ClassVar[dict[type[Service], Callable[[], Service]]] = {}
+def get_service(service_type: type[T]) -> T:
+    return Container(_registry).get(service_type)
 
-    @classmethod
-    def register(
-        cls,
-        service_type: type[Service],
-        service_factory: Callable[[], Service],
-    ) -> None:
-        """
-        Register a service by providing its class
-        and the service factory callable.
-        """
-        cls._registry[service_type] = service_factory
 
-    @classmethod
-    def get(cls, service_type: type[Service]) -> Service:
-        """
-        Get a service by its class.
-        """
-        service_factory = cls._registry.get(service_type)
-
-        if not service_factory:
-            raise ServiceTypeNotRegisteredError
-
-        return service_factory()
+async def aget_service(service_type: type[T]) -> T:
+    return await Container(_registry).aget(service_type)

tests/example_app/__init__.py

@@ -1,35 +1,54 @@
 from pathlib import Path
-from typing import Annotated, cast
+from typing import Annotated
 
 from fastapi import APIRouter, Depends
 from typer import Typer, echo
 
-from wheke import Pod, Service, ServiceRegistry, Wheke
+from wheke import Pod, Wheke, get_service
+from wheke.service import aget_service
 
 STATIC_PATH = Path(__file__).parent / "static"
 
 router = APIRouter()
 cli = Typer()
 
 
-class TestService(Service):
+class PingService:
     def ping(self) -> str:
         return "pong"
 
 
-def test_service_factory() -> TestService:
-    return TestService()
+def ping_service_factory() -> PingService:
+    return PingService()
 
 
-def get_test_service() -> TestService:
-    return cast(TestService, ServiceRegistry.get(TestService))
+def get_ping_service() -> PingService:
+    return get_service(PingService)
+
+
+class APingService:
+    async def ping(self) -> str:
+        return "pong"
+
+
+async def aping_service_factory() -> APingService:
+    return APingService()
+
+
+async def get_aping_service() -> APingService:
+    return await aget_service(APingService)
 
 
 @router.get("/ping")
-def ping(service: Annotated[TestService, Depends(get_test_service)]) -> dict:
+def ping(service: Annotated[PingService, Depends(get_ping_service)]) -> dict:
     return {"value": service.ping()}
 
 
+@router.get("/aping")
+async def aping(service: Annotated[APingService, Depends(get_aping_service)]) -> dict:
+    return {"value": await service.ping()}
+
+
 @cli.callback()
 def callback() -> None:
     pass
@@ -45,7 +64,10 @@ def hello() -> None:
     router=router,
     static_url="/static",
     static_path=str(STATIC_PATH),
-    services=[(TestService, test_service_factory)],
+    services=[
+        (PingService, ping_service_factory),
+        (APingService, aping_service_factory),
+    ],
     cli=cli,
 )
 

tests/test_app.py

@@ -51,3 +51,10 @@ def test_ping(client: TestClient) -> None:
 
     assert response.status_code == status.HTTP_200_OK
     assert response.json() == {"value": "pong"}
+
+
+def test_aping(client: TestClient) -> None:
+    response = client.get("/aping")
+
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json() == {"value": "pong"}