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 

49@overload 

50def safe( 

51 func: Callable[P, T], 

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

53 

54 

55@overload 

56def safe( 

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

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

59 

60 

61def safe( 

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

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

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

65 

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

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

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

69 

70 Args: 

71 func: The sync or async function to wrap. 

72 

73 Returns: 

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

75 (or a coroutine resolving to one). 

76 

77 Example: 

78 >>> @safe 

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

80 ... return int(s) 

81 >>> parse_int("42") 

82 Ok(42) 

83 >>> parse_int("invalid") 

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

85 

86 """ 

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

88 # (LOAD_DEREF is faster than LOAD_GLOBAL) 

89 ok_cls = Ok 

90 err_cls = Err 

91 

92 if inspect.iscoroutinefunction(func): 

93 # Cast once here to satisfy mypy inside the closure 

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

95 

96 @functools.wraps(func) 

97 async def async_wrapper( 

98 *args: P.args, 

99 **kwargs: P.kwargs, 

100 ) -> Result[T, Exception]: 

101 try: 

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

103 except Exception as e: 

104 return err_cls(e) 

105 

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

107 

108 # Cast once here to satisfy mypy inside the closure 

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

110 

111 @functools.wraps(func) 

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

113 try: 

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

115 except Exception as e: 

116 return err_cls(e) 

117 

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

119 

120 

121class SafeFromDecorator(Protocol[E_co]): 

122 """Protocol for safe_from decorator.""" 

123 

124 @overload 

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

126 

127 @overload 

128 def __call__( 

129 self, 

130 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_iterable( 

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

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

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

200 ok_cls = Ok 

201 err_cls = Err 

202 values: list[T] = [] 

203 append = values.append 

204 for result in results: 

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

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

207 if isinstance(result, ok_cls): 

208 append(result.ok_value) 

209 elif isinstance(result, err_cls): 

210 return result 

211 else: 

212 # Fallback for structural compatibility 

213 return result # type: ignore[unreachable] 

214 return ok_cls(values) 

215 

216 

217def collect_results( 

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

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

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

221 

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

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

224 

225 Args: 

226 results: Iterable of Result objects. 

227 

228 Returns: 

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

230 

231 Example: 

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

233 Ok([1, 2, 3]) 

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

235 Err("fail") 

236 

237 """ 

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

239 optimized_res = _collect_list(results) 

240 if optimized_res is not None: 

241 return optimized_res 

242 

243 return _collect_iterable(results) 

244 

245 

246@overload 

247async def map_async( 

248 result: Ok[T], 

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

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

251 

252 

253@overload 

254async def map_async( 

255 result: Err[E], 

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

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

258 

259 

260@overload 

261async def map_async( 

262 result: Result[T, E], 

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

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

265 

266 

267async def map_async( 

268 result: Result[T, E], 

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

270) -> Result[U, E]: 

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

272 

273 If the result is Err, returns it unchanged. 

274 

275 Args: 

276 result: The Result to process. 

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

278 

279 Returns: 

280 New Result containing the processed value or original error. 

281 

282 Example: 

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

284 ... return str(x * 2) 

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

286 Ok('10') 

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

288 Err('fail') 

289 

290 """ 

291 if isinstance(result, Ok): 

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

293 return result 

294 

295 

296@overload 

297async def and_then_async( 

298 result: Ok[T], 

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

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

301 

302 

303@overload 

304async def and_then_async( 

305 result: Err[E], 

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

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

308 

309 

310@overload 

311async def and_then_async( 

312 result: Result[T, E], 

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

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

315 

316 

317async def and_then_async( 

318 result: Result[T, E], 

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

320) -> Result[U, E]: 

321 """Asynchronously chain operations that return Results. 

322 

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

324 If the result is Err, returns it unchanged. 

325 

326 Args: 

327 result: The Result to process. 

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

329 

330 Returns: 

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

332 

333 Example: 

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

335 ... if uid == 1: 

336 ... return Ok("Alice") 

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

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

339 Ok('Alice') 

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

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

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

343 Err(ValueError('No DB')) 

344 

345 """ 

346 if isinstance(result, Ok): 

347 return await func(result.ok_value) 

348 return result