Coverage for greyhorse / river / private / monads / reader_t.py: 94%

1# mypy: disable_error_code="misc,type-arg,valid-type,override,index,union-attr,attr-defined,arg-type" 

2from __future__ import annotations 

3 

4from collections.abc import Callable 

5 

6from greyhorse.utils.types import TypeWrapper 

7 

8from .effect import Effect, ExitCase 

9 

10 

11class ReaderT[R, F, T](Effect[T], TypeWrapper[R, F, T]): 

12 """Reader monad transformer — lifts an Effect into a Reader context. 

13 

14 Wraps ``R -> F[T]`` where ``F`` is an Effect type (typically ``IO``). 

15 ``ResourceReader`` extends this with ``Resource[IO, T]`` as the inner Effect. 

16 

17 Type parameters: 

18 R: Environment (reader context) type. 

19 F: Inner Effect type. 

20 T: Value type. 

21 """ 

22 

23 __slots__ = ('_run',) 

24 

25 def __init__(self, run: Callable[[R], F[T]]) -> None: 

26 self._run = run 

27 

28 def map[U](self, f: Callable[[F[T]], F[U]]) -> ReaderT[R, F, U]: 

29 """Apply a function to the inner Effect.""" 

30 def run(r: R) -> F[U]: 

31 a = self._run(r) 

32 return f(a) 

33 

34 return ReaderT(run) 

35 

36 def and_then[U](self, f: Callable[[F[T]], ReaderT[R, F, U]]) -> ReaderT[R, F, U]: 

37 """Monadic bind — chain with a function returning ReaderT.""" 

38 def run(r: R) -> F[U]: 

39 a = self._run(r) 

40 return f(a)(r) 

41 

42 return ReaderT(run) 

43 

44 def __call__(self, r: R) -> F[T]: 

45 """Execute with the given environment, returning the inner Effect.""" 

46 return self._run(r) 

47 

48 @classmethod 

49 def pure(cls, value: T) -> ReaderT[R, F, T]: 

50 """Wrap a pure value into ReaderT using the inner Effect's ``pure``.""" 

51 return ReaderT(lambda _: cls.__wrapped_type__[1].pure(value)) 

52 

53 @classmethod 

54 def ask(cls) -> ReaderT[R, F, T]: 

55 """Return the environment as the value.""" 

56 return ReaderT(cls.__wrapped_type__[1].pure) 

57 

58 def bracket_case[U]( # type: ignore 

59 self, 

60 use: Callable[[T], ReaderT[R, F, U]], 

61 release: Callable[[T, ExitCase], ReaderT[R, F, None]], 

62 ) -> ReaderT[R, F, U]: 

63 """Acquire-use-release with ExitCase, lifted into ReaderT. 

64 

65 Args: 

66 use: Function applied to the acquired value. 

67 release: Cleanup function with ExitCase feedback. 

68 """ 

69 def run(r: R) -> U: 

70 wrapped_acquire = self._run(r) 

71 

72 def wrapped_release(v: T, ec: ExitCase) -> F[None]: 

73 return release(v, ec)(r) 

74 

75 def wrapped_use(v: T) -> F[U]: 

76 return use(v)(r) 

77 

78 return wrapped_acquire.bracket_case(wrapped_use, wrapped_release) 

79 

80 return ReaderT(run)