Coverage for src / taipanstack / core / result.py: 100%

1""" 

2Result type utilities for functional error handling. 

3 

4Provides Rust-style Result types (Ok/Err) for explicit error handling, 

5avoiding exceptions for expected failure cases. This promotes safer, 

6more predictable code. 

7 

8Example: 

9 >>> from taipanstack.core.result import safe, Ok, Err 

10 >>> @safe 

11 ... def divide(a: int, b: int) -> float: 

12 ... if b == 0: 

13 ... raise ValueError("division by zero") 

14 ... return a / b 

15 >>> result = divide(10, 0) 

16 >>> match result: 

17 ... case Err(e): 

18 ... print(f"Error: {e}") 

19 ... case Ok(value): 

20 ... print(f"Result: {value}") 

21 Error: division by zero 

22 

23""" 

24 

25import functools 

26import inspect 

27from collections.abc import Awaitable, Callable, Iterable 

28from typing import ParamSpec, Protocol, TypeVar, cast, overload 

29 

30from result import Err, Ok, Result 

31 

32__all__ = [ 

33 "Err", 

34 "Ok", 

35 "Result", 

36 "and_then_async", 

37 "collect_results", 

38 "map_async", 

39 "safe", 

40 "safe_from", 

41] 

42 

43P = ParamSpec("P") 

44T = TypeVar("T") 

45E = TypeVar("E", bound=Exception) 

46E_co = TypeVar("E_co", bound=Exception, covariant=True) 

47U = TypeVar("U") 

48 

49 

50@overload 

51def safe( 

52 func: Callable[P, T], 

53) -> Callable[P, Result[T, Exception]]: ... 

54 

55 

56@overload 

57def safe( 

58 func: Callable[P, Awaitable[T]], 

59) -> Callable[P, Awaitable[Result[T, Exception]]]: ... 

60 

61 

62def safe( 

63 func: Callable[P, T] | Callable[P, Awaitable[T]], 

64) -> Callable[P, Result[T, Exception]] | Callable[P, Awaitable[Result[T, Exception]]]: 

65 """Wrap a sync or async function to convert exceptions into Err results. 

66 

67 Detect whether *func* is a coroutine function and choose the 

68 appropriate wrapper so that ``await``-able functions remain 

69 ``await``-able and synchronous functions stay synchronous. 

70 

71 Args: 

72 func: The sync or async function to wrap. 

73 

74 Returns: 

75 A wrapped function that returns ``Result[T, Exception]`` 

76 (or a coroutine resolving to one). 

77 

78 Example: 

79 >>> @safe 

80 ... def parse_int(s: str) -> int: 

81 ... return int(s) 

82 >>> parse_int("42") 

83 Ok(42) 

84 >>> parse_int("invalid") 

85 Err(ValueError("invalid literal for int()...")) 

86 

87 """ 

88 # Pre-cache constructors for minor speedup in tight loops 

89 # (LOAD_DEREF is faster than LOAD_GLOBAL) 

90 ok_cls = Ok 

91 err_cls = Err 

92 

93 if inspect.iscoroutinefunction(func): 

94 # Cast once here to satisfy mypy inside the closure 

95 func_coro = cast(Callable[P, Awaitable[T]], func) 

96 

97 @functools.wraps(func) 

98 async def async_wrapper( 

99 *args: P.args, 

100 **kwargs: P.kwargs, 

101 ) -> Result[T, Exception]: 

102 try: 

103 return ok_cls(await func_coro(*args, **kwargs)) 

104 except Exception as e: 

105 return err_cls(e) 

106 

107 return cast(Callable[P, Awaitable[Result[T, Exception]]], async_wrapper) 

108 

109 # Cast once here to satisfy mypy inside the closure 

110 func_sync = cast(Callable[P, T], func) 

111 

112 @functools.wraps(func) 

113 def wrapper(*args: P.args, **kwargs: P.kwargs) -> Result[T, Exception]: 

114 try: 

115 return ok_cls(func_sync(*args, **kwargs)) 

116 except Exception as e: 

117 return err_cls(e) 

118 

119 return cast(Callable[P, Result[T, Exception]], wrapper) 

120 

121 

122class SafeFromDecorator(Protocol[E_co]): 

123 """Protocol for safe_from decorator.""" 

124 

125 @overload 

126 def __call__(self, func: Callable[P, T]) -> Callable[P, Result[T, E_co]]: ... 

127 

128 @overload 

129 def __call__( 

130 self, func: Callable[P, Awaitable[T]] 

131 ) -> Callable[P, Awaitable[Result[T, E_co]]]: ... 

132 

133 

134def safe_from( 

135 *exception_types: type[E], 

136) -> SafeFromDecorator[E]: 

137 """Decorator factory to catch specific exceptions as Err. 

138 

139 Only catches specified exception types; others propagate normally. 

140 

141 Args: 

142 *exception_types: Exception types to convert to Err. 

143 

144 Returns: 

145 Decorator that wraps function with selective error handling. 

146 

147 Example: 

148 >>> @safe_from(ValueError, TypeError) 

149 ... def process(data: str) -> int: 

150 ... return int(data) 

151 >>> process("abc") 

152 Err(ValueError(...)) 

153 

154 """ 

155 

156 def decorator( 

157 func: Callable[P, T] | Callable[P, Awaitable[T]], 

158 ) -> Callable[P, Result[T, E]] | Callable[P, Awaitable[Result[T, E]]]: 

159 if inspect.iscoroutinefunction(func): 

160 func_coro = cast(Callable[P, Awaitable[T]], func) 

161 

162 @functools.wraps(func) 

163 async def async_wrapper(*args: P.args, **kwargs: P.kwargs) -> Result[T, E]: 

164 try: 

165 return Ok(await func_coro(*args, **kwargs)) 

166 except exception_types as e: 

167 return Err(e) 

168 

169 return cast(Callable[P, Awaitable[Result[T, E]]], async_wrapper) 

170 

171 func_sync = cast(Callable[P, T], func) 

172 

173 @functools.wraps(func) 

174 def wrapper(*args: P.args, **kwargs: P.kwargs) -> Result[T, E]: 

175 try: 

176 return Ok(func_sync(*args, **kwargs)) 

177 except exception_types as e: 

178 return Err(e) 

179 

180 return cast(Callable[P, Result[T, E]], wrapper) 

181 

182 return cast(SafeFromDecorator[E], decorator) 

183 

184 

185def _collect_list( 

186 results: list[Result[T, E]] | tuple[Result[T, E], ...], 

187) -> Result[list[T], E] | None: 

188 try: 

189 # We use a runtime # type: ignore to bypass mypy's strict check 

190 # on the AttributeError strategy for extreme performance on the hot path 

191 return Ok([r.ok_value for r in results]) # type: ignore[union-attr] 

192 except AttributeError: 

193 return None 

194 

195 

196def collect_results( 

197 results: Iterable[Result[T, E]], 

198) -> Result[list[T], E]: 

199 """Collect an iterable of Results into a single Result. 

200 

201 If all results are Ok, returns Ok with list of values. 

202 If any result is Err, returns the first Err encountered. 

203 

204 Args: 

205 results: Iterable of Result objects. 

206 

207 Returns: 

208 Ok(list[T]) if all are Ok, otherwise first Err. 

209 

210 Example: 

211 >>> collect_results([Ok(1), Ok(2), Ok(3)]) 

212 Ok([1, 2, 3]) 

213 >>> collect_results([Ok(1), Err("fail"), Ok(3)]) 

214 Err("fail") 

215 

216 """ 

217 if isinstance(results, (list, tuple)): 

218 optimized_res = _collect_list(results) 

219 if optimized_res is not None: 

220 return optimized_res 

221 

222 ok_cls = Ok 

223 err_cls = Err 

224 values: list[T] = [] 

225 append = values.append 

226 for result in results: 

227 # We use explicit type checks (isinstance) but pre-cache the constructors. 

228 # type(result) is ok_cls would be even faster but breaks subclassing. 

229 if isinstance(result, ok_cls): 

230 append(result.ok_value) 

231 elif isinstance(result, err_cls): 

232 return result 

233 else: 

234 # Fallback for structural compatibility 

235 return result 

236 return ok_cls(values) 

237 

238 

239@overload 

240async def map_async( 

241 result: Ok[T], 

242 func: Callable[[T], Awaitable[U]], 

243) -> Result[U, E]: ... 

244 

245 

246@overload 

247async def map_async( 

248 result: Err[E], 

249 func: Callable[[T], Awaitable[U]], 

250) -> Err[E]: ... 

251 

252 

253@overload 

254async def map_async( 

255 result: Result[T, E], 

256 func: Callable[[T], Awaitable[U]], 

257) -> Result[U, E]: ... 

258 

259 

260async def map_async( 

261 result: Result[T, E], 

262 func: Callable[[T], Awaitable[U]], 

263) -> Result[U, E]: 

264 """Asynchronously apply a function to the value of an Ok result. 

265 

266 If the result is Err, returns it unchanged. 

267 

268 Args: 

269 result: The Result to process. 

270 func: Awaitable function to apply to the Ok value. 

271 

272 Returns: 

273 New Result containing the processed value or original error. 

274 

275 Example: 

276 >>> async def process(x: int) -> str: 

277 ... return str(x * 2) 

278 >>> await map_async(Ok(5), process) 

279 Ok('10') 

280 >>> await map_async(Err("fail"), process) 

281 Err('fail') 

282 

283 """ 

284 match result: 

285 case Ok(val): 

286 return Ok(await func(val)) 

287 case Err(e): 

288 _ = e 

289 return result 

290 case _: 

291 return result 

292 

293 

294@overload 

295async def and_then_async( 

296 result: Ok[T], 

297 func: Callable[[T], Awaitable[Result[U, E]]], 

298) -> Result[U, E]: ... 

299 

300 

301@overload 

302async def and_then_async( 

303 result: Err[E], 

304 func: Callable[[T], Awaitable[Result[U, E]]], 

305) -> Err[E]: ... 

306 

307 

308@overload 

309async def and_then_async( 

310 result: Result[T, E], 

311 func: Callable[[T], Awaitable[Result[U, E]]], 

312) -> Result[U, E]: ... 

313 

314 

315async def and_then_async( 

316 result: Result[T, E], 

317 func: Callable[[T], Awaitable[Result[U, E]]], 

318) -> Result[U, E]: 

319 """Asynchronously chain operations that return Results. 

320 

321 If the result is Ok, asynchronously applies `func` and returns its Result. 

322 If the result is Err, returns it unchanged. 

323 

324 Args: 

325 result: The Result to process. 

326 func: Awaitable function taking the Ok value and returning a new Result. 

327 

328 Returns: 

329 The new Result from `func` or the original error. 

330 

331 Example: 

332 >>> async def fetch_user(uid: int) -> Result[str, ValueError]: 

333 ... if uid == 1: 

334 ... return Ok("Alice") 

335 ... return Err(ValueError("User not found")) 

336 >>> await and_then_async(Ok(1), fetch_user) 

337 Ok('Alice') 

338 >>> await and_then_async(Ok(2), fetch_user) 

339 Err(ValueError('User not found')) 

340 >>> await and_then_async(Err(ValueError("No DB")), fetch_user) 

341 Err(ValueError('No DB')) 

342 

343 """ 

344 match result: 

345 case Ok(val): 

346 return await func(val) 

347 case Err(e): 

348 _ = e 

349 return result 

350 case _: 

351 return result