Coverage for src/monadc/try_/success.py: 100%

1from typing import Callable, Union, Any, Optional 

2from .try_ import Try, T, U 

3 

4 

5class Success(Try[T]): 

6 """ 

7 Represents a successful computation in a Try monad. 

8 

9 Success contains the result of a computation that completed without throwing an exception. 

10 """ 

11 

12 __match_args__ = ("_value",) 

13 

14 def __new__(cls, value: T) -> 'Success[T]': 

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

16 return object.__new__(cls) 

17 

18 def __init__(self, value: T) -> None: 

19 # If _value 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, '_value'): 

22 return 

23 

24 self._value = value 

25 

26 def is_success(self) -> bool: 

27 return True 

28 

29 def is_failure(self) -> bool: 

30 return False 

31 

32 def get(self) -> T: 

33 return self._value 

34 

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

36 return self._value 

37 

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

39 return None 

40 

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

42 try: 

43 result = func(self._value) 

44 return Success(result) 

45 except Exception as e: 

46 from .failure import Failure 

47 return Failure(e) 

48 

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

50 try: 

51 return func(self._value) 

52 except Exception as e: 

53 from .failure import Failure 

54 return Failure(e) 

55 

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

57 try: 

58 if predicate(self._value): 

59 return self 

60 else: 

61 from .failure import Failure 

62 return Failure(ValueError("Predicate does not hold")) 

63 except Exception as e: 

64 from .failure import Failure 

65 return Failure(e) 

66 

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

68 # Success doesn't need recovery 

69 return self 

70 

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

72 # Success doesn't need recovery 

73 return self 

74 

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

76 return if_success(self._value) 

77 

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

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

80 try: 

81 return success_func(self._value) 

82 except Exception as e: 

83 from .failure import Failure 

84 return Failure(e) 

85 

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

87 func(self._value) 

88 

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

90 """Return self since Success doesn't need alternative.""" 

91 return self 

92 

93 def flatten(self) -> Any: # Returns Try[U] where T = Try[U] 

94 """Flatten Success[Try[U]] to Try[U]. Raises TypeError if T is not Try[U].""" 

95 if isinstance(self._value, Try): 

96 return self._value 

97 else: 

98 # Following Scala's behavior: throw if trying to flatten non-nested Try 

99 raise TypeError(f"Cannot flatten Success[{type(self._value).__name__}] - value must be a Try instance") 

100 

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

102 """Safe flatten that returns self if not nested (idempotent behavior).""" 

103 if isinstance(self._value, Try): 

104 return self._value 

105 else: 

106 return self 

107 

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

109 from ..option import Option 

110 return Option(self._value) 

111 

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

113 from ..either import Right 

114 return Right(self._value) 

115 

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

117 if isinstance(other, Success): 

118 return self._value == other._value # type: ignore[no-any-return] 

119 return False 

120 

121 def __repr__(self) -> str: 

122 return f"Success({self._value!r})" 

123 

124 def __str__(self) -> str: 

125 return self.__repr__()