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

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

2from __future__ import annotations 

3 

4from collections.abc import Callable 

5 

6from greyhorse.result import Err, Ok, Result 

7 

8 

9class RWSR[R, X, S, T, E]: 

10 """Reader + Writer + State + Result monad — the full stack. 

11 

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

13 

14 Type parameters: 

15 R: Environment type. 

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

17 S: State type. 

18 T: Success value type. 

19 E: Error type. 

20 """ 

21 

22 __slots__ = ('_run',) 

23 

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

25 self._run = run 

26 

27 def map[U](self, f: Callable[[T], U]) -> RWSR[R, X, S, U, E]: 

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

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

30 a, s, x = self._run(r, s) 

31 return a.map(f), s, x 

32 

33 return RWSR(run) 

34 

35 def and_then[U](self, f: Callable[[T], RWSR[R, X, S, U, E]]) -> RWSR[R, X, S, U, E]: 

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

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

38 a, s1, x1 = self._run(r, s) 

39 match a: 

40 case Ok(v): 

41 b, s2, x2 = f(v)(r, s1) 

42 return b, s2, x1 + x2 

43 case Err(_): 

44 return a, s1, x1 

45 

46 return RWSR(run) 

47 

48 @staticmethod 

49 def tell(x: X) -> RWSR[R, X, S, None, E]: 

50 """Append to the writer output.""" 

51 return RWSR(lambda r, s: (Ok(), s, x)) 

52 

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

54 """Execute the computation with environment and initial state.""" 

55 return self._run(r, state) 

56 

57 @staticmethod 

58 def pure(value: Result[T, E], monoid: X) -> RWSR[R, X, S, T, E]: 

59 """Wrap a Result value with initial writer output.""" 

60 return RWSR(lambda r, s: (value, s, monoid)) 

61 

62 @staticmethod 

63 def get(monoid: X) -> RWSR[R, X, S, T, E]: 

64 """Return the current state as value.""" 

65 return RWSR(lambda r, s: (s, s, monoid)) 

66 

67 @staticmethod 

68 def put(s: S, monoid: X) -> RWSR[R, X, S, None, E]: 

69 """Replace the current state.""" 

70 return RWSR(lambda r, _: (Ok(), s, monoid))