Selector provider (#32)

* Selector provider

Author: jasonprado

docs/providers/selector.md

@@ -0,0 +1,26 @@
+# Selector
+
+- Selector provider chooses between a provider based on a key.
+- Resolves into a single dependency.
+
+```python
+import os
+from typing import Protocol
+from that_depends import BaseContainer, providers
+
+class StorageService(Protocol):
+    ...
+
+class StorageServiceLocal(StorageService):
+    ...
+
+class StorageServiceRemote(StorageService):
+    ...
+
+class DIContainer(BaseContainer):
+    storage_service = providers.Selector(
+        lambda: os.getenv("STORAGE_BACKEND", "local"),
+        local=providers.Factory(StorageServiceLocal),
+        remote=providers.Factory(StorageServiceRemote),
+    )
+```

tests/container.py

@@ -8,6 +8,8 @@
 
 logger = logging.getLogger(__name__)
 
+global_state_for_selector: typing.Literal["sync_resource", "async_resource", "missing"] = "sync_resource"
+
 
 def create_sync_resource() -> typing.Iterator[datetime.datetime]:
     logger.debug("Resource initiated")
@@ -57,6 +59,11 @@ class DIContainer(BaseContainer):
     sync_resource = providers.Resource(create_sync_resource)
     async_resource = providers.AsyncResource(create_async_resource)
     sequence = providers.List(sync_resource, async_resource)
+    selector: providers.Selector[datetime.datetime] = providers.Selector(
+        lambda: global_state_for_selector,
+        sync_resource=sync_resource,
+        async_resource=async_resource,
+    )
 
     simple_factory = providers.Factory(SimpleFactory, dep1="text", dep2=123)
     async_factory = providers.AsyncFactory(async_factory, async_resource.cast)

tests/test_main_providers.py

@@ -51,6 +51,34 @@ async def test_list_provider() -> None:
     assert sequence == [sync_resource, async_resource]
 
 
+async def test_selector_provider_async() -> None:
+    container.global_state_for_selector = "async_resource"
+    selected = await DIContainer.selector()
+    async_resource = await DIContainer.async_resource()
+
+    assert selected == async_resource
+
+
+async def test_selector_provider_async_missing() -> None:
+    container.global_state_for_selector = "missing"
+    with pytest.raises(RuntimeError):
+        await DIContainer.selector()
+
+
+async def test_selector_provider_sync() -> None:
+    container.global_state_for_selector = "sync_resource"
+    selected = DIContainer.selector.sync_resolve()
+    sync_resource = DIContainer.sync_resource.sync_resolve()
+
+    assert selected == sync_resource
+
+
+async def test_selector_provider_sync_missing() -> None:
+    container.global_state_for_selector = "missing"
+    with pytest.raises(RuntimeError):
+        DIContainer.selector.sync_resolve()
+
+
 async def test_singleton_provider() -> None:
     singleton1 = await DIContainer.singleton()
     singleton2 = await DIContainer.singleton()

that_depends/providers/__init__.py

@@ -8,6 +8,7 @@
 )
 from that_depends.providers.factories import AsyncFactory, Factory
 from that_depends.providers.resources import AsyncResource, Resource
+from that_depends.providers.selector import Selector
 from that_depends.providers.singleton import Singleton
 
 
@@ -22,6 +23,7 @@
     "Factory",
     "List",
     "Resource",
+    "Selector",
     "Singleton",
     "container_context",
 ]

that_depends/providers/selector.py

@@ -0,0 +1,26 @@
+import typing
+
+from that_depends.providers.base import AbstractProvider
+
+
+T = typing.TypeVar("T")
+
+
+class Selector(AbstractProvider[T]):
+    def __init__(self, selector: typing.Callable[[], str], **providers: AbstractProvider[T]) -> None:
+        self._selector = selector
+        self._providers = providers
+
+    async def async_resolve(self) -> T:
+        selected_key = self._selector()
+        if selected_key not in self._providers:
+            msg = f"No provider matches {selected_key}"
+            raise RuntimeError(msg)
+        return await self._providers[selected_key].async_resolve()
+
+    def sync_resolve(self) -> T:
+        selected_key = self._selector()
+        if selected_key not in self._providers:
+            msg = f"No provider matches {selected_key}"
+            raise RuntimeError(msg)
+        return self._providers[selected_key].sync_resolve()