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

1# mypy: disable_error_code="misc,arg-type" 

2"""Effect protocol and ExitCase enum — foundation of the monadic algebra. 

3 

4``Effect[T]`` is the protocol for effectful computations with bracket semantics. 

5``ExitCase`` communicates how a bracketed use-phase ended. 

6""" 

7 

8from __future__ import annotations 

9 

10import asyncio 

11from collections.abc import Callable 

12from typing import Protocol 

13 

14from greyhorse.enum import Enum, Struct, Unit 

15 

16 

17class ExitCase(Enum): 

18 """Outcome of a bracketed computation. 

19 

20 Used by ``bracket_case`` to communicate how the ``use`` phase ended, 

21 so the release function can decide cleanup strategy. 

22 

23 Variants: 

24 Succeeded: The use-phase completed normally. 

25 Canceled: The use-phase was canceled (``asyncio.CancelledError``). 

26 Errored(exc): The use-phase raised an exception. 

27 

28 Examples: 

29 :: 

30 

31 ExitCase.from_exc(None) # Succeeded 

32 ExitCase.from_exc(asyncio.CancelledError()) # Canceled 

33 ExitCase.from_exc(ValueError('x')) # Errored(exc=ValueError('x')) 

34 

35 Pattern matching:: 

36 

37 match exit_case: 

38 case ExitCase.Succeeded: 

39 commit() 

40 case ExitCase.Errored(exc=e): 

41 rollback(e) 

42 case ExitCase.Canceled: 

43 rollback(None) 

44 """ 

45 

46 Succeeded = Unit() 

47 Canceled = Unit() 

48 Errored = Struct(exc=Exception) 

49 

50 @staticmethod 

51 def from_exc(exc: BaseException | None) -> ExitCase: 

52 """Create an ExitCase from an exception (or None for success). 

53 

54 Args: 

55 exc: The exception, or None if the computation succeeded. 

56 

57 Returns: 

58 ``Succeeded`` if exc is None, ``Canceled`` for ``CancelledError``, 

59 ``Errored(exc=...)`` otherwise. 

60 """ 

61 if exc is None: 

62 return ExitCase.Succeeded 

63 if isinstance(exc, asyncio.CancelledError): 

64 return ExitCase.Canceled 

65 return ExitCase.Errored(exc=exc) 

66 

67 

68class Effect[T](Protocol): 

69 """Protocol for effectful computations with bracket semantics. 

70 

71 Implementations must provide ``pure``, ``and_then``, and ``bracket_case``. 

72 ``IO`` is the canonical synchronous implementation. 

73 

74 Type parameters: 

75 T: The result type of the computation. 

76 """ 

77 

78 @staticmethod 

79 def pure(value: T) -> Effect[T]: 

80 """Wrap a pure value into this Effect (no side effects).""" 

81 ... 

82 

83 def and_then[U](self, f: Callable[[T], Effect[U]]) -> Effect[U]: 

84 """Monadic bind — chain with a function returning Effect.""" 

85 ... 

86 

87 def bracket[U]( 

88 self, use: Callable[[T], Effect[U]], release: Callable[[T], Effect[None]] 

89 ) -> Effect[U]: 

90 """Acquire-use-release with simple release (no ExitCase info). 

91 

92 Args: 

93 use: Function applied to the acquired value. 

94 release: Cleanup function (called regardless of outcome). 

95 

96 Returns: 

97 Effect that runs acquire → use → release. 

98 """ 

99 return self.bracket_case(use, lambda v, _: release(v)) 

100 

101 def bracket_case[U]( 

102 self, use: Callable[[T], Effect[U]], release: Callable[[T, ExitCase], Effect[None]] 

103 ) -> Effect[U]: 

104 """Acquire-use-release with ExitCase feedback to the release function. 

105 

106 Args: 

107 use: Function applied to the acquired value. 

108 release: Cleanup function ``(value, exit_case) -> Effect[None]``. 

109 

110 Returns: 

111 Effect that runs acquire → use → release with ExitCase. 

112 """ 

113 ...