feat: add Runtime.use_effect (#5)

Author: suned

README.md

@@ -345,6 +345,73 @@ type Try[E, R] = Effect[Never, E, R]
 ```
 For effects that do not require abilities, but might fail.
 
+Sometimes, instantiating abilities may itself require side-effects. For example, consider a program that requires a `Config` ability:
+
+
+```python
+from stateless import Depend
+
+
+class Config:
+    ...
+
+
+def main() -> Depend[Config, None]:
+    ...
+```
+
+Now imagine that you want to provide the `Config` ability by reading from environment variables:
+
+
+```python
+import os
+
+from stateless import Depend, depend
+
+
+class OS:
+    environ: dict[str, str] = os.environ
+
+
+def get_config() -> Depend[OS, Config]:
+    os = yield from depend(OS)
+    return Config(
+        url=os.environ['AUTH_TOKEN'],
+        auth_token=os.environ['URL']
+    )
+```
+
+To supply the `Config` instance returned from `get_config`, we can use `Runtime.use_effect`:
+
+
+```python
+from stateless import Runtime
+
+
+Runtime().use(OS()).use_effect(get_config()).run(main())
+```
+
+`Runtime.use_effect` assumes that all abilities required by the effect given as its argument can be provided by the runtime. If this is not the case, you'll get a type-checker error:
+
+```python
+from stateless import Depend, Runtime
+
+
+class A:
+    pass
+
+
+class B:
+    pass
+
+
+def get_B() -> Depend[A, B]:
+    ...
+
+Runtime().use(A()).use_effect(get_B())  # OK
+Runtime().use_effect(get_B())           # Type-checker error!
+```
+
 ## Error Handling
 
 So far we haven't used the error type `E` for anything: We've simply parameterized it with `typing.Never`. We've claimed that this means that the effect doesn't fail. This is of course not literally true, as exceptions can still occur even if we parameterize `E` with `Never.`

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "stateless"
-version = "0.5.1"
+version = "0.5.2"
 description = "Statically typed, purely functional effects for Python"
 authors = ["suned <[email protected]>"]
 readme = "README.md"

src/stateless/effect.py

@@ -1,6 +1,6 @@
 """Contains the Effect type and core functions for working with effects."""
 
-from collections.abc import Generator
+from collections.abc import Generator, Hashable
 from dataclasses import dataclass, field
 from functools import lru_cache, partial, wraps
 from types import TracebackType
@@ -9,7 +9,7 @@
 from typing_extensions import Never, ParamSpec, TypeAlias
 
 R = TypeVar("R")
-A = TypeVar("A")
+A = TypeVar("A", bound=Hashable)
 E = TypeVar("E", bound=Exception)
 P = ParamSpec("P")
 E2 = TypeVar("E2", bound=Exception)

src/stateless/runtime.py

@@ -1,5 +1,6 @@
 """Runtime for executing effects."""
 
+from collections.abc import Generator, Hashable
 from dataclasses import dataclass
 from functools import cache
 from typing import Generic, Literal, Tuple, Type, TypeVar, cast, overload
@@ -8,9 +9,9 @@
 from stateless.errors import MissingAbilityError
 from stateless.parallel import Parallel
 
-A = TypeVar("A")
-A2 = TypeVar("A2")
-A3 = TypeVar("A3")
+A = TypeVar("A", bound=Hashable)
+A2 = TypeVar("A2", bound=Hashable)
+A3 = TypeVar("A3", bound=Hashable)
 R = TypeVar("R")
 E = TypeVar("E", bound=Exception)
 
@@ -23,11 +24,14 @@ def _get_ability(ability_type: Type[A], abilities: Tuple[A, ...]) -> A:
     raise MissingAbilityError(ability_type)
 
 
-@dataclass(frozen=True)
+@dataclass(frozen=True, init=False)
 class Runtime(Generic[A]):
     """A runtime for executing effects."""
 
-    abilities: tuple[A, ...] = ()
+    abilities: tuple[A, ...]
+
+    def __init__(self, *abilities: A):
+        object.__setattr__(self, "abilities", abilities)
 
     def use(self, ability: A2) -> "Runtime[A | A2]":
         """
@@ -43,7 +47,25 @@ def use(self, ability: A2) -> "Runtime[A | A2]":
         -------
             A new runtime with the ability.
         """
-        return Runtime((ability, *self.abilities))
+        return Runtime(*(*self.abilities, ability))  # type: ignore
+
+    def use_effect(self, effect: Effect[A, Exception, A2]) -> "Runtime[A | A2]":
+        """
+        Use an ability produced by an effect with this runtime.
+
+        Enables running effects that require the ability.
+
+        All abilities required by `effect` must be provided by the runtime.
+
+        Args:
+        ----
+            effect: The effect producing an ability.
+
+        Returns:
+        -------
+            A new runtime with the ability.
+        """
+        return self.use(effect)  # type: ignore
 
     def get_ability(self, ability_type: Type[A]) -> A:
         """
@@ -61,12 +83,18 @@ def get_ability(self, ability_type: Type[A]) -> A:
         return _get_ability(ability_type, self.abilities)  # type: ignore
 
     @overload
-    def run(self, effect: Effect[A, E, R], return_errors: Literal[False] = False) -> R:
+    def run(
+        self,
+        effect: Effect[A, E, R],
+        return_errors: Literal[False] = False,
+    ) -> R:
         ...  # pragma: no cover
 
     @overload
     def run(
-        self, effect: Effect[A, E, R], return_errors: Literal[True] = True
+        self,
+        effect: Effect[A, E, R],
+        return_errors: Literal[True] = True,
     ) -> R | E:
         ...  # pragma: no cover
 
@@ -83,6 +111,20 @@ def run(self, effect: Effect[A, E, R], return_errors: bool = False) -> R | E:
         -------
             The result of the effect.
         """
+        abilities: tuple[A, ...] = ()
+        for ability in self.abilities:
+            if isinstance(ability, Generator):
+                abilities = (  # pyright: ignore
+                    self._run(ability, abilities, return_errors=False),
+                    *abilities,
+                )
+            else:
+                abilities = (ability, *abilities)  # pyright: ignore
+        return self._run(effect, abilities, return_errors)
+
+    def _run(
+        self, effect: Effect[A, E, R], abilities: Tuple[A, ...], return_errors: bool
+    ) -> R | E:
         try:
             ability_or_error = next(effect)
 
@@ -101,7 +143,7 @@ def run(self, effect: Effect[A, E, R], return_errors: bool = False) -> R | E:
                         case ability_type if ability_type is Parallel:
                             ability_or_error = effect.send(self)
                         case ability_type:
-                            ability = self.get_ability(ability_type)
+                            ability = _get_ability(ability_type, abilities)
                             ability_or_error = effect.send(ability)
                 except MissingAbilityError as error:
                     ability_or_error = effect.throw(error)

tests/test_runtime.py

@@ -46,13 +46,11 @@ def effect() -> Depend[Super, Super]:
     with raises(MissingAbilityError, match="Super") as info:
         Runtime().run(effect())  # type: ignore
 
-    print(info.getrepr())
-
-    # test that the third frame is the yield
+    # test that the fourth frame is the yield
     # expression in `effect` function above
     # (first is Runtime().run(..)
     # second is effect.throw in Runtime.run)
-    frame = info.traceback[2]
+    frame = info.traceback[3]
     assert str(frame.path) == __file__
     assert frame.lineno == effect.__code__.co_firstlineno
 
@@ -98,3 +96,11 @@ def catches() -> Try[ValueError, None]:
 
     with raises(ValueError, match="oops again"):
         Runtime().run(catches(), return_errors=True)
+
+
+def test_use_effect() -> None:
+    def effect() -> Depend[str, bytes]:
+        ability: str = yield str
+        return ability.encode()
+
+    assert Runtime("ability").use_effect(effect()).run(depend(bytes)) == b"ability"