Coverage for greyhorse / river / private / monads / rwr.py: 100%

1# mypy: warn_no_return=false,disable_error_code="operator,return-value" 

2from __future__ import annotations 

3 

4from collections.abc import Callable 

5 

6from greyhorse.result import Err, Ok, Result 

7 

8 

9class RWR[R, X, T, E]: 

10 """Reader + Writer + Result monad — environment, output accumulation, and failure. 

11 

12 Wraps ``R -> (Result[T, E], X)``. Short-circuits on ``Err``. 

13 

14 Type parameters: 

15 R: Environment type. 

16 X: Writer output type (must support ``+``). 

17 T: Success value type. 

18 E: Error type. 

19 """ 

20 

21 __slots__ = ('_run',) 

22 

23 def __init__(self, run: Callable[[R], tuple[Result[T, E], X]]) -> None: 

24 self._run = run 

25 

26 def map[U](self, f: Callable[[T], U]) -> RWR[R, X, U, E]: 

27 """Apply a pure function to the success value.""" 

28 def run(r: R) -> tuple[Result[U, E], X]: 

29 a, x = self._run(r) 

30 return a.map(f), x 

31 

32 return RWR(run) 

33 

34 def and_then[U](self, f: Callable[[T], RWR[R, X, U, E]]) -> RWR[R, X, U, E]: 

35 """Monadic bind — chain on success, concatenate outputs.""" 

36 def run(r: R) -> tuple[Result[U, E], X]: 

37 a, x1 = self._run(r) 

38 match a: 

39 case Ok(v): 

40 b, x2 = f(v)(r) 

41 return b, x1 + x2 

42 case Err(_): 

43 return a, x1 

44 

45 return RWR(run) 

46 

47 def __call__(self, r: R) -> tuple[Result[T, E], X]: 

48 """Execute the computation with the given environment.""" 

49 return self._run(r)