Coverage for src/monadc/try_/failure.py: 97%

1from typing import Callable, Union, Any, Optional, cast 

2from .try_ import Try, T, U 

3 

4 

5class Failure(Try[T]): 

6 """ 

7 Represents a failed computation in a Try monad. 

8 

9 Failure contains the exception that was thrown during computation. 

10 """ 

11 

12 __match_args__ = ("_exception",) 

13 

14 def __new__(cls, exception: Exception) -> 'Failure[T]': 

15 """Create a new Failure instance directly, bypassing Try.__new__.""" 

16 return object.__new__(cls) 

17 

18 def __init__(self, exception: Exception) -> None: 

19 # If _exception already exists, this is a second call from Python's object creation process 

20 # We should ignore it since the object has already been properly initialized 

21 if hasattr(self, '_exception'): 

22 return 

23 

24 self._exception = exception 

25 

26 def is_success(self) -> bool: 

27 return False 

28 

29 def is_failure(self) -> bool: 

30 return True 

31 

32 def __bool__(self) -> bool: 

33 """Failure is falsy.""" 

34 return False 

35 

36 def get(self) -> T: 

37 # Re-raise the original exception 

38 raise self._exception 

39 

40 def get_or_else(self, default: Union[T, Callable[[], T]]) -> T: 

41 if callable(default): 

42 try: 

43 return default() 

44 except Exception: 

45 # If default function also fails, re-raise original exception 

46 raise self._exception 

47 return default 

48 

49 def exception(self) -> Optional[Exception]: 

50 return self._exception 

51 

52 def map(self, func: Callable[[T], U]) -> Try[U]: 

53 # Failure passes through unchanged 

54 return cast(Try[U], self) 

55 

56 def flat_map(self, func: Callable[[T], Try[U]]) -> Try[U]: 

57 # Failure passes through unchanged 

58 return cast(Try[U], self) 

59 

60 def filter(self, predicate: Callable[[T], bool]) -> Try[T]: 

61 # Failure passes through unchanged 

62 return self 

63 

64 def recover(self, func: Callable[[Exception], T]) -> Try[T]: 

65 try: 

66 result = func(self._exception) 

67 from .success import Success 

68 return Success(result) 

69 except Exception as e: 

70 return Failure(e) 

71 

72 def recover_with(self, func: Callable[[Exception], Try[T]]) -> Try[T]: 

73 try: 

74 return func(self._exception) 

75 except Exception as e: 

76 return Failure(e) 

77 

78 def fold(self, if_failure: Callable[[Exception], U], if_success: Callable[[T], U]) -> U: 

79 return if_failure(self._exception) 

80 

81 def transform(self, success_func: Callable[[T], Try[U]], 

82 failure_func: Callable[[Exception], Try[U]]) -> Try[U]: 

83 try: 

84 return failure_func(self._exception) 

85 except Exception as e: 

86 return Failure(e) 

87 

88 def foreach(self, func: Callable[[T], Any]) -> None: 

89 # Failure does nothing 

90 pass 

91 

92 def or_else(self, alternative: Union[Try[T], Callable[[], Try[T]]]) -> Try[T]: 

93 """Return alternative since this is Failure.""" 

94 if callable(alternative): 

95 try: 

96 return alternative() 

97 except Exception as e: 

98 return Failure(e) 

99 return alternative 

100 

101 def flatten(self) -> Any: # Returns Try[U] 

102 """Flatten Failure to itself since Failure has no inner Try to unwrap.""" 

103 return self 

104 

105 def flatten_safe(self) -> Any: # Returns Try[U] 

106 """Safe flatten that returns self (same as flatten for Failure).""" 

107 return self 

108 

109 def to_option(self) -> Any: # Returns Option[T] 

110 from ..option import Nil 

111 return Nil() 

112 

113 def to_either(self) -> Any: # Returns Either[Exception, T] 

114 from ..either import Left 

115 return Left(self._exception) 

116 

117 def __eq__(self, other: object) -> bool: 

118 if isinstance(other, Failure): 

119 # Compare exceptions by type and message 

120 return (type(self._exception) == type(other._exception) and 

121 str(self._exception) == str(other._exception)) 

122 return False 

123 

124 def __repr__(self) -> str: 

125 return f"Failure({self._exception!r})" 

126 

127 def __str__(self) -> str: 

128 return self.__repr__()