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 >>> if isinstance(result, Err): 

17 ... print(f"Error: {result.err_value}") 

18 ... else: 

19 ... print(f"Result: {result.unwrap()}") 

20 Error: division by zero 

21 

22""" 

23 

24import functools 

25import inspect 

26from collections.abc import Awaitable, Callable, Iterable 

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

28 

29from result import Err, Ok, Result 

30 

31__all__ = [ 

32 "Err", 

33 "Ok", 

34 "Result", 

35 "and_then_async", 

36 "collect_results", 

37 "map_async", 

38 "safe", 

39 "safe_from", 

40] 

41 

42P = ParamSpec("P") 

43T = TypeVar("T") 

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

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

46U = TypeVar("U") 

47 

48 

49def _safe_async_wrapper( 

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

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

52 ok_cls = Ok 

53 err_cls = Err 

54 

55 @functools.wraps(func) 

56 async def async_wrapper( 

57 *args: P.args, 

58 **kwargs: P.kwargs, 

59 ) -> Result[T, Exception]: 

60 try: 

61 return ok_cls(await func(*args, **kwargs)) 

62 except Exception as e: 

63 return err_cls(e) 

64 

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

66 

67 

68def _safe_sync_wrapper( 

69 func: Callable[P, T], 

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

71 ok_cls = Ok 

72 err_cls = Err 

73 

74 @functools.wraps(func) 

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

76 try: 

77 return ok_cls(func(*args, **kwargs)) 

78 except Exception as e: 

79 return err_cls(e) 

80 

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

82 

83 

84@overload 

85def safe( 

86 func: Callable[P, T], 

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

88 

89 

90@overload 

91def safe( 

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

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

94 

95 

96def safe( 

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

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

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

100 

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

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

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

104 

105 Args: 

106 func: The sync or async function to wrap. 

107 

108 Returns: 

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

110 (or a coroutine resolving to one). 

111 

112 Example: 

113 >>> @safe 

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

115 ... return int(s) 

116 >>> parse_int("42") 

117 Ok(42) 

118 >>> parse_int("invalid") 

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

120 

121 """ 

122 if inspect.iscoroutinefunction(func): 

123 return _safe_async_wrapper(cast(Callable[P, Awaitable[T]], func)) 

124 return _safe_sync_wrapper(cast(Callable[P, T], func)) 

125 

126 

127class SafeFromDecorator(Protocol[E_co]): 

128 """Protocol for safe_from decorator.""" 

129 

130 @overload 

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

132 

133 @overload 

134 def __call__( 

135 self, 

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

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

138 

139 

140def _safe_from_async_wrapper( 

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

142 exception_types: tuple[type[E], ...], 

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

144 

145 @functools.wraps(func) 

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

147 try: 

148 return Ok(await func(*args, **kwargs)) 

149 except exception_types as e: 

150 return Err(e) 

151 

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

153 

154 

155def _safe_from_sync_wrapper( 

156 func: Callable[P, T], 

157 exception_types: tuple[type[E], ...], 

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

159 

160 @functools.wraps(func) 

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

162 try: 

163 return Ok(func(*args, **kwargs)) 

164 except exception_types as e: 

165 return Err(e) 

166 

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

168 

169 

170def safe_from( 

171 *exception_types: type[E], 

172) -> SafeFromDecorator[E]: 

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

174 

175 Only catches specified exception types; others propagate normally. 

176 

177 Args: 

178 *exception_types: Exception types to convert to Err. 

179 

180 Returns: 

181 Decorator that wraps function with selective error handling. 

182 

183 Example: 

184 >>> @safe_from(ValueError, TypeError) 

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

186 ... return int(data) 

187 >>> process("abc") 

188 Err(ValueError(...)) 

189 

190 """ 

191 

192 def decorator( 

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

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

195 if inspect.iscoroutinefunction(func): 

196 return _safe_from_async_wrapper( 

197 cast(Callable[P, Awaitable[T]], func), exception_types 

198 ) 

199 return _safe_from_sync_wrapper(cast(Callable[P, T], func), exception_types) 

200 

201 return cast(SafeFromDecorator[E], decorator) 

202 

203 

204def _extract_ok_values( 

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

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

207 try: 

208 return Ok([cast(Ok[T], r).ok_value for r in results]) 

209 except AttributeError: 

210 return None 

211 

212 

213def _collect_list( 

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

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

216 # Explicitly verify types to avoid bypassing formal proof 

217 # We fall back to the slow path if we aren't absolutely certain 

218 # it's a structural match, to handle edge cases properly. 

219 if not results: 

220 return Ok([]) 

221 

222 first = results[0] 

223 # Fast path: homogeneous exact types 

224 if type(first) is Ok: 

225 return _extract_ok_values(results) 

226 

227 return None 

228 

229 

230def _collect_iterable( 

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

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

233 """Collect an iterable of Results iteratively.""" 

234 ok_cls = Ok 

235 err_cls = Err 

236 values: list[T] = [] 

237 append = values.append 

238 for result in results: 

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

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

241 if isinstance(result, ok_cls): 

242 append(result.ok_value) 

243 elif isinstance(result, err_cls): 

244 return result 

245 else: 

246 raise TypeError(f"Expected Result, got {type(result)}") 

247 return ok_cls(values) 

248 

249 

250def collect_results( 

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

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

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

254 

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

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

257 

258 Args: 

259 results: Iterable of Result objects. 

260 

261 Returns: 

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

263 

264 Example: 

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

266 Ok([1, 2, 3]) 

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

268 Err("fail") 

269 

270 """ 

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

272 res_seq = cast(list[Result[T, E]] | tuple[Result[T, E], ...], results) 

273 optimized_res = _collect_list(res_seq) 

274 if optimized_res is not None: 

275 return optimized_res 

276 

277 return _collect_iterable(results) 

278 

279 

280@overload 

281async def map_async( 

282 result: Ok[T], 

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

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

285 

286 

287@overload 

288async def map_async( 

289 result: Err[E], 

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

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

292 

293 

294@overload 

295async def map_async( 

296 result: Result[T, E], 

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

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

299 

300 

301async def map_async( 

302 result: Result[T, E], 

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

304) -> Result[U, E]: 

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

306 

307 If the result is Err, returns it unchanged. 

308 

309 Args: 

310 result: The Result to process. 

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

312 

313 Returns: 

314 New Result containing the processed value or original error. 

315 

316 Example: 

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

318 ... return str(x * 2) 

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

320 Ok('10') 

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

322 Err('fail') 

323 

324 """ 

325 if isinstance(result, Ok): 

326 return Ok(await func(result.ok_value)) 

327 return result 

328 

329 

330@overload 

331async def and_then_async( 

332 result: Ok[T], 

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

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

335 

336 

337@overload 

338async def and_then_async( 

339 result: Err[E], 

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

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

342 

343 

344@overload 

345async def and_then_async( 

346 result: Result[T, E], 

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

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

349 

350 

351async def and_then_async( 

352 result: Result[T, E], 

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

354) -> Result[U, E]: 

355 """Asynchronously chain operations that return Results. 

356 

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

358 If the result is Err, returns it unchanged. 

359 

360 Args: 

361 result: The Result to process. 

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

363 

364 Returns: 

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

366 

367 Example: 

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

369 ... if uid == 1: 

370 ... return Ok("Alice") 

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

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

373 Ok('Alice') 

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

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

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

377 Err(ValueError('No DB')) 

378 

379 """ 

380 if isinstance(result, Ok): 

381 return await func(result.ok_value) 

382 return result