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 Callable, Coroutine, Iterable 

28from typing import Any, 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, Coroutine[Any, Any, T]], 

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

60 

61 

62def safe( 

63 func: Callable[P, T] | Callable[P, Coroutine[Any, Any, T]], 

64) -> ( 

65 Callable[P, Result[T, Exception]] 

66 | Callable[P, Coroutine[Any, Any, Result[T, Exception]]] 

67): 

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

69 

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

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

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

73 

74 Args: 

75 func: The sync or async function to wrap. 

76 

77 Returns: 

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

79 (or a coroutine resolving to one). 

80 

81 Example: 

82 >>> @safe 

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

84 ... return int(s) 

85 >>> parse_int("42") 

86 Ok(42) 

87 >>> parse_int("invalid") 

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

89 

90 """ 

91 if inspect.iscoroutinefunction(func): 

92 

93 @functools.wraps(func) 

94 async def async_wrapper( 

95 *args: P.args, 

96 **kwargs: P.kwargs, 

97 ) -> Result[T, Exception]: 

98 try: 

99 func_coro = cast(Callable[P, Coroutine[Any, Any, T]], func) 

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

101 except Exception as e: 

102 return Err(e) 

103 

104 return cast( 

105 Callable[P, Coroutine[Any, Any, Result[T, Exception]]], async_wrapper 

106 ) 

107 

108 @functools.wraps(func) 

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

110 try: 

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

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

113 except Exception as e: 

114 return Err(e) 

115 

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

117 

118 

119class SafeFromDecorator(Protocol[E_co]): 

120 """Protocol for safe_from decorator.""" 

121 

122 @overload 

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

124 

125 @overload 

126 def __call__( 

127 self, func: Callable[P, Coroutine[Any, Any, T]] 

128 ) -> Callable[P, Coroutine[Any, Any, Result[T, E_co]]]: ... 

129 

130 

131def safe_from( 

132 *exception_types: type[E], 

133) -> SafeFromDecorator[E]: 

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

135 

136 Only catches specified exception types; others propagate normally. 

137 

138 Args: 

139 *exception_types: Exception types to convert to Err. 

140 

141 Returns: 

142 Decorator that wraps function with selective error handling. 

143 

144 Example: 

145 >>> @safe_from(ValueError, TypeError) 

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

147 ... return int(data) 

148 >>> process("abc") 

149 Err(ValueError(...)) 

150 

151 """ 

152 

153 def decorator( 

154 func: Callable[P, T] | Callable[P, Coroutine[Any, Any, T]], 

155 ) -> Callable[P, Result[T, E]] | Callable[P, Coroutine[Any, Any, Result[T, E]]]: 

156 if inspect.iscoroutinefunction(func): 

157 

158 @functools.wraps(func) 

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

160 try: 

161 func_coro = cast(Callable[P, Coroutine[Any, Any, T]], func) 

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

163 except exception_types as e: 

164 return Err(e) 

165 

166 return cast(Callable[P, Coroutine[Any, Any, Result[T, E]]], async_wrapper) 

167 

168 @functools.wraps(func) 

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

170 try: 

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

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

173 except exception_types as e: 

174 return Err(e) 

175 

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

177 

178 return cast(SafeFromDecorator[E], decorator) 

179 

180 

181def collect_results( 

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

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

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

185 

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

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

188 

189 Args: 

190 results: Iterable of Result objects. 

191 

192 Returns: 

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

194 

195 Example: 

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

197 Ok([1, 2, 3]) 

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

199 Err("fail") 

200 

201 """ 

202 values: list[T] = [] 

203 append = values.append 

204 for result in results: 

205 try: 

206 append(result.ok_value) # type: ignore[union-attr] 

207 except AttributeError: 

208 return result # type: ignore[return-value] 

209 return Ok(values) 

210 

211 

212@overload 

213async def map_async( 

214 result: Ok[T], 

215 func: Callable[[T], Coroutine[Any, Any, U]], 

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

217 

218 

219@overload 

220async def map_async( 

221 result: Err[E], 

222 func: Callable[[T], Coroutine[Any, Any, U]], 

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

224 

225 

226@overload 

227async def map_async( 

228 result: Result[T, E], 

229 func: Callable[[T], Coroutine[Any, Any, U]], 

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

231 

232 

233async def map_async( 

234 result: Result[T, E], 

235 func: Callable[[T], Coroutine[Any, Any, U]], 

236) -> Result[U, E]: 

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

238 

239 If the result is Err, returns it unchanged. 

240 

241 Args: 

242 result: The Result to process. 

243 func: Coroutine function to apply to the Ok value. 

244 

245 Returns: 

246 New Result containing the processed value or original error. 

247 

248 Example: 

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

250 ... return str(x * 2) 

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

252 Ok('10') 

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

254 Err('fail') 

255 

256 """ 

257 try: 

258 val = result.ok_value # type: ignore[union-attr] 

259 except AttributeError: 

260 return result # type: ignore[return-value] 

261 return Ok(await func(val)) 

262 

263 

264@overload 

265async def and_then_async( 

266 result: Ok[T], 

267 func: Callable[[T], Coroutine[Any, Any, Result[U, E]]], 

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

269 

270 

271@overload 

272async def and_then_async( 

273 result: Err[E], 

274 func: Callable[[T], Coroutine[Any, Any, Result[U, E]]], 

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

276 

277 

278@overload 

279async def and_then_async( 

280 result: Result[T, E], 

281 func: Callable[[T], Coroutine[Any, Any, Result[U, E]]], 

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

283 

284 

285async def and_then_async( 

286 result: Result[T, E], 

287 func: Callable[[T], Coroutine[Any, Any, Result[U, E]]], 

288) -> Result[U, E]: 

289 """Asynchronously chain operations that return Results. 

290 

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

292 If the result is Err, returns it unchanged. 

293 

294 Args: 

295 result: The Result to process. 

296 func: Coroutine function taking the Ok value and returning a new Result. 

297 

298 Returns: 

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

300 

301 Example: 

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

303 ... if uid == 1: 

304 ... return Ok("Alice") 

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

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

307 Ok('Alice') 

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

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

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

311 Err(ValueError('No DB')) 

312 

313 """ 

314 try: 

315 val = result.ok_value # type: ignore[union-attr] 

316 except AttributeError: 

317 return result # type: ignore[return-value] 

318 return await func(val)