Coverage for src / lexigram / contracts / core / result.py: 5%

1"""Result type — concrete Ok/Err implementation. 

2 

3Canonical location for Result[T, E], Ok, and Err. 

4All Lexigram packages import from here. 

5""" 

6 

7from __future__ import annotations 

8 

9from collections.abc import Awaitable, Callable 

10from typing import Any, Generic, TypeVar, cast 

11 

12T = TypeVar("T") 

13E = TypeVar("E") 

14U = TypeVar("U") 

15F = TypeVar("F") 

16 

17 

18class UnwrapError(Exception): 

19 """Raised when unwrap() or unwrap_err() is called on the wrong variant. 

20 

21 Provides a clearer error type than the generic ``ValueError`` so callers 

22 can catch it explicitly when needed. 

23 """ 

24 

25 _code: str = "LEX_ERR_RESULT_002" 

26 

27 def __init__(self, message: str, result: Any | None = None) -> None: 

28 """Initialise the error. 

29 

30 Args: 

31 message: Human-readable description of what was attempted. 

32 result: The ``Result`` instance that caused the error (optional). 

33 """ 

34 super().__init__(message) 

35 self.result = result 

36 

37 

38class Result(Generic[T, E]): 

39 """Base Result type. Not abstract — Ok and Err are the only variants.""" 

40 

41 __slots__ = () 

42 

43 def is_ok(self) -> bool: 

44 raise NotImplementedError 

45 

46 def is_err(self) -> bool: 

47 raise NotImplementedError 

48 

49 def unwrap(self) -> T: 

50 raise NotImplementedError 

51 

52 def unwrap_err(self) -> E: 

53 raise NotImplementedError 

54 

55 def unwrap_or(self, default: T) -> T: 

56 raise NotImplementedError 

57 

58 def unwrap_or_else(self, op: Callable[[E], T]) -> T: 

59 raise NotImplementedError 

60 

61 def map_sync(self, op: Callable[[T], U]) -> Result[U, E]: 

62 raise NotImplementedError 

63 

64 def map_err(self, op: Callable[[E], F]) -> Result[T, F]: 

65 raise NotImplementedError 

66 

67 def and_then_sync(self, op: Callable[[T], Result[U, E]]) -> Result[U, E]: 

68 raise NotImplementedError 

69 

70 def or_else_sync(self, op: Callable[[E], Result[T, F]]) -> Result[T, F]: 

71 raise NotImplementedError 

72 

73 def expect(self, message: str) -> T: 

74 raise NotImplementedError 

75 

76 def match(self, ok: Callable[[T], U], err: Callable[[E], U]) -> U: 

77 raise NotImplementedError 

78 

79 async def map(self, op: Callable[[T], Awaitable[U]]) -> Result[U, E]: 

80 raise NotImplementedError 

81 

82 async def and_then( 

83 self, op: Callable[[T], Awaitable[Result[U, E]]] 

84 ) -> Result[U, E]: 

85 raise NotImplementedError 

86 

87 async def or_else(self, op: Callable[[E], Awaitable[Result[T, F]]]) -> Result[T, F]: 

88 raise NotImplementedError 

89 

90 def flatten(self) -> Result[Any, E]: 

91 raise NotImplementedError 

92 

93 def filter(self, predicate: Callable[[T], bool], error: E) -> Result[T, E]: 

94 raise NotImplementedError 

95 

96 def ok_or(self, default: U) -> T | U: 

97 raise NotImplementedError 

98 

99 @classmethod 

100 def from_exception( 

101 cls, 

102 exc: Exception, 

103 ok_type: type[T] = type(None), # type: ignore[assignment] 

104 ) -> Result[T, Exception]: 

105 """Wrap a caught exception into an Err result.""" 

106 return Err(exc) 

107 

108 def to_optional(self) -> T | None: 

109 return self.unwrap() if self.is_ok() else None 

110 

111 def inspect(self, op: Callable[[T], None]) -> Result[T, E]: 

112 if self.is_ok(): 

113 op(self.unwrap()) 

114 return self 

115 

116 def inspect_err(self, op: Callable[[E], None]) -> Result[T, E]: 

117 if self.is_err(): 

118 op(self.unwrap_err()) 

119 return self 

120 

121 

122class Ok(Result[T, E]): 

123 __slots__ = ("_value",) 

124 __match_args__ = ("_value",) 

125 

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

127 self._value = value 

128 

129 def is_ok(self) -> bool: 

130 return True 

131 

132 def is_err(self) -> bool: 

133 return False 

134 

135 def unwrap(self) -> T: 

136 return self._value 

137 

138 def unwrap_err(self) -> E: 

139 raise UnwrapError(f"Called unwrap_err on Ok({self._value!r})") 

140 

141 def unwrap_or(self, default: T) -> T: 

142 return self._value 

143 

144 def unwrap_or_else(self, op: Callable[[E], T]) -> T: 

145 return self._value 

146 

147 def map_sync(self, op: Callable[[T], U]) -> Result[U, E]: 

148 return Ok(op(self._value)) 

149 

150 def map_err(self, op: Callable[[E], F]) -> Result[T, F]: 

151 return cast("Result[T, F]", self) 

152 

153 def and_then_sync(self, op: Callable[[T], Result[U, E]]) -> Result[U, E]: 

154 return op(self._value) 

155 

156 def or_else_sync(self, op: Callable[[E], Result[T, F]]) -> Result[T, F]: 

157 return cast("Result[T, F]", self) 

158 

159 def expect(self, message: str) -> T: 

160 return self._value 

161 

162 def match(self, ok: Callable[[T], U], err: Callable[[E], U]) -> U: 

163 return ok(self._value) 

164 

165 async def map(self, op: Callable[[T], Awaitable[U]]) -> Result[U, E]: 

166 return Ok(await op(self._value)) 

167 

168 async def async_map(self, op: Callable[[T], Awaitable[U]]) -> Result[U, E]: 

169 """Alias for ``map`` — exists for backward compatibility.""" 

170 return Ok(await op(self._value)) 

171 

172 async def and_then( 

173 self, op: Callable[[T], Awaitable[Result[U, E]]] 

174 ) -> Result[U, E]: 

175 return await op(self._value) 

176 

177 async def or_else(self, op: Callable[[E], Awaitable[Result[T, F]]]) -> Result[T, F]: 

178 return cast("Result[T, F]", self) 

179 

180 def flatten(self) -> Result[Any, E]: 

181 if isinstance(self._value, Result): 

182 return self._value 

183 return cast("Result[Any, E]", self) 

184 

185 def filter(self, predicate: Callable[[T], bool], error: E) -> Result[T, E]: 

186 return self if predicate(self._value) else Err(error) 

187 

188 def ok_or(self, default: U) -> T: 

189 return self._value 

190 

191 def __repr__(self) -> str: 

192 return f"Ok({self._value!r})" 

193 

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

195 return isinstance(other, Ok) and self._value == other._value 

196 

197 def __hash__(self) -> int: 

198 return hash(("Ok", self._value)) 

199 

200 

201class Err(Result[T, E]): 

202 __slots__ = ("_error",) 

203 __match_args__ = ("_error",) 

204 

205 def __init__(self, error: E) -> None: 

206 self._error = error 

207 

208 def is_ok(self) -> bool: 

209 return False 

210 

211 def is_err(self) -> bool: 

212 return True 

213 

214 def unwrap(self) -> T: 

215 raise UnwrapError(f"Called unwrap on Err({self._error!r})", self) 

216 

217 def unwrap_err(self) -> E: 

218 return self._error 

219 

220 def unwrap_or(self, default: T) -> T: 

221 return default 

222 

223 def unwrap_or_else(self, op: Callable[[E], T]) -> T: 

224 return op(self._error) 

225 

226 def map_sync(self, op: Callable[[T], U]) -> Result[U, E]: 

227 return cast("Result[U, E]", self) 

228 

229 def map_err(self, op: Callable[[E], F]) -> Result[T, F]: 

230 return Err(op(self._error)) 

231 

232 def and_then_sync(self, op: Callable[[T], Result[U, E]]) -> Result[U, E]: 

233 return cast("Result[U, E]", self) 

234 

235 def or_else_sync(self, op: Callable[[E], Result[T, F]]) -> Result[T, F]: 

236 return op(self._error) 

237 

238 def expect(self, message: str) -> T: 

239 raise UnwrapError(f"{message}: {self._error!r}", self) 

240 

241 def match(self, ok: Callable[[T], U], err: Callable[[E], U]) -> U: 

242 return err(self._error) 

243 

244 async def map(self, op: Callable[[T], Awaitable[U]]) -> Result[U, E]: 

245 return cast("Result[U, E]", self) 

246 

247 async def async_map(self, op: Callable[[T], Awaitable[U]]) -> Result[U, E]: 

248 """Alias for ``map`` — exists for backward compatibility.""" 

249 return cast("Result[U, E]", self) 

250 

251 async def and_then( 

252 self, op: Callable[[T], Awaitable[Result[U, E]]] 

253 ) -> Result[U, E]: 

254 return cast("Result[U, E]", self) 

255 

256 async def or_else(self, op: Callable[[E], Awaitable[Result[T, F]]]) -> Result[T, F]: 

257 return await op(self._error) 

258 

259 def flatten(self) -> Result[Any, E]: 

260 return cast("Result[Any, E]", self) 

261 

262 def filter(self, predicate: Callable[[T], bool], error: E) -> Result[T, E]: 

263 return self 

264 

265 def ok_or(self, default: U) -> U: 

266 return default 

267 

268 def __repr__(self) -> str: 

269 return f"Err({self._error!r})" 

270 

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

272 return isinstance(other, Err) and self._error == other._error 

273 

274 def __hash__(self) -> int: 

275 return hash(("Err", self._error)) 

276 

277 

278__all__ = ["Err", "Ok", "Result", "UnwrapError"]