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

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

2from __future__ import annotations 

3 

4from collections.abc import Callable 

5 

6from greyhorse.result import Err, Ok, Result 

7 

8 

9class RR[R, T, E]: 

10 """Reader + Result monad — computation that depends on environment and can fail. 

11 

12 Wraps a function ``R -> Result[T, E]``. Short-circuits on ``Err``. 

13 

14 Type parameters: 

15 R: Environment type. 

16 T: Success value type. 

17 E: Error type. 

18 

19 Examples: 

20 :: 

21 

22 rr = RR(lambda r: Ok(r + 1)) 

23 assert rr(10).unwrap() == 11 

24 

25 # Err short-circuits the chain 

26 err_rr = RR(lambda r: Err('fail')) 

27 chained = err_rr.and_then(lambda v: RR(lambda r: Ok(v * 2))) 

28 assert chained(5).is_err() 

29 """ 

30 

31 __slots__ = ('_run',) 

32 

33 def __init__(self, function: Callable[[R], Result[T, E]]) -> None: 

34 self._run = function 

35 

36 def map[U](self, f: Callable[[T], U]) -> RR[R, U, E]: 

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

38 def run(r: R) -> Result[U, E]: 

39 a = self._run(r) 

40 return a.map(f) 

41 

42 return RR(run) 

43 

44 def and_then[U](self, f: Callable[[T], RR[R, U, E]]) -> RR[R, U, E]: 

45 """Monadic bind — chain on success, short-circuit on error.""" 

46 def run(r: R) -> Result[U, E]: 

47 a = self._run(r) 

48 match a: 

49 case Ok(v): 

50 return f(v)(r) 

51 case Err(_): 

52 return a 

53 

54 return RR(run) 

55 

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

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

58 return self._run(r)