Coverage for src / ocarina / dsl / testing_with_railway / internals / action_chain.py: 53.16%

1"""Railway Oriented Programming implementation for action chains. 

2 

3Implements the Railway Oriented Programming (ROP) pattern for building 

4resilient, composable action chains with explicit error handling. 

5 

6The pattern provides: 

7- Type-safe success/failure representation (Result[T] = Ok[T] | Fail) 

8- Builder pattern for configuring error and success handlers 

9- Automatic short-circuiting on failure (railway switches tracks) 

10- Composable action chains with .then() combinator 

11 

12Railway metaphor: 

13 Code as railway track with two rails: 

14 - Success rail (top): Actions execute normally → Ok results 

15 - Failure rail (bottom): Action failed → Fail results 

16 

17 When an action fails, the train switches to the failure rail and 

18 stays there (short-circuits). Subsequent actions become no-ops. 

19 

20Core types: 

21 Result[T] → Ok[T] | Fail (from railway.result) 

22 Action[T] → Thunk producing Result[T] 

23 ActionChain → Railway track with current position 

24 

25State machine: 

26 ActionStart → .failure() → ActionFailure → .success() → ActionSuccess 

27 → .execute() → ActionChain 

28 → .then() 

29 → ... 

30 

31Example: 

32 >>> def risky_action() -> Result[int]: 

33 ... if random.random() > 0.5: 

34 ... return Ok(42) 

35 ... return Fail(Exception("Bad luck")) 

36 ... 

37 >>> chain = ( 

38 ... ActionStart(risky_action) 

39 ... .failure(lambda e: logger.error(f"Failed: {e}")) 

40 ... .success(lambda: logger.success("Succeeded")) 

41 ... .execute() 

42 ... ) 

43 ... 

44 >>> if chain.has_failed(): 

45 ... logger.warning("On failure rail") 

46 

47""" 

48 

49from collections.abc import Callable 

50from typing import final 

51 

52from ocarina.custom_types.effect import Effect 

53from ocarina.custom_types.thunk import Thunk 

54from ocarina.railway.result import Result, is_fail 

55 

56type Action[T] = Thunk[Result[T]] 

57"""Deferred computation producing a Result when called. 

58 

59Signature: () -> Result[T] 

60 

61Example: 

62 >>> def create_action() -> Action[int]: 

63 ... return lambda: Ok(expensive_computation()) 

64""" 

65 

66type FailureHandler = Callable[[Exception], None] 

67"""Handler for failures, receives the error. 

68 

69Used for logging, screenshots, cleanup, metrics. 

70 

71Example: 

72 >>> def handle_error(exc: Exception) -> None: 

73 ... logger.error(f"Failed: {exc}", exc=exc) 

74 ... take_screenshot(driver, "ERROR") 

75""" 

76 

77type SuccHandler = Effect 

78"""Handler for successes (Effect). 

79 

80Used for logging, screenshots, metrics. 

81 

82Example: 

83 >>> def handle_success() -> None: 

84 ... logger.success("Action succeeded") 

85""" 

86 

87 

88@final 

89class ActionStart[T]: 

90 """Initial action chain builder state, awaiting failure handler. 

91 

92 Entry point for building an action chain. 

93 

94 Example: 

95 >>> ActionStart(action).failure(handle_error).success(handle_success).execute() 

96 

97 """ 

98 

99 def __init__(self, action: Action[T]) -> None: 

100 """Initialize with action to execute.""" 

101 self.__action__ = action 

102 

103 def failure(self, failure_handler: FailureHandler) -> ActionFailure[T]: 

104 """Configure error handler and advance to next state.""" 

105 return ActionFailure(self.__action__, failure_handler) 

106 

107 

108@final 

109class ActionFailure[T]: 

110 """Action builder with failure handler configured, awaiting success handler.""" 

111 

112 def __init__(self, action: Action[T], failure_handler: FailureHandler) -> None: 

113 """Initialize with action and failure handler.""" 

114 self._action = action 

115 self._failure_handler = failure_handler 

116 

117 def success(self, success_handler: SuccHandler) -> ActionSuccess[T]: 

118 """Configure success handler and advance to executable state.""" 

119 return ActionSuccess(self._action, self._failure_handler, success_handler) 

120 

121 

122@final 

123class ActionSuccess[T]: 

124 """Fully configured action builder, ready for execution.""" 

125 

126 def __init__( 

127 self, 

128 action: Action[T], 

129 failure_handler: FailureHandler, 

130 success_handler: SuccHandler, 

131 ) -> None: 

132 """Initialize with action and both handlers.""" 

133 self.__action__ = action 

134 self.__failure_handler__ = failure_handler 

135 self.__success_handler__ = success_handler 

136 

137 def execute(self) -> ActionChain[T]: 

138 """Execute action and call appropriate handler based on result. 

139 

140 Flow: 

141 1. Call action() to get Result[T] 

142 2. If Fail: call failure_handler(error) 

143 3. If Ok: call success_handler() 

144 4. Return ActionChain with result 

145 

146 Example: 

147 >>> chain = ( 

148 ... ActionStart(lambda: Ok(42)) 

149 ... .failure(lambda e: print(f"Error: {e}")) 

150 ... .success(lambda: print("Success")) 

151 ... .execute() 

152 ... ) 

153 

154 """ 

155 result = self.__action__() 

156 

157 if is_fail(result): 

158 self.__failure_handler__(result.error) 

159 return ActionChain(has_failed=True, result=result) 

160 

161 self.__success_handler__() 

162 return ActionChain(has_failed=False, result=result) 

163 

164 

165@final 

166class NeutralActionStart[T]: 

167 """Neutral action builder that short-circuits (on failure rail). 

168 

169 When a previous action failed, subsequent actions become "neutral" - 

170 they accept the same builder API but don't execute. This implements 

171 railway short-circuiting while maintaining fluent API. 

172 """ 

173 

174 def __init__(self, *, result: Result[T] | None) -> None: 

175 """Initialize with failure result from previous action.""" 

176 self._result = result 

177 

178 def failure(self, *args, **kwargs) -> NeutralActionFailure[T]: # noqa: ARG002 

179 """Accept failure handler (ignored) and continue short-circuit.""" 

180 return NeutralActionFailure(result=self._result) 

181 

182 

183@final 

184class NeutralActionFailure[T]: 

185 """Neutral action builder in failure-configured state (no-op).""" 

186 

187 def __init__(self, *, result: Result[T] | None) -> None: 

188 """Initialize with failure result.""" 

189 self._result = result 

190 

191 def success(self, *args, **kwargs) -> NeutralActionSuccess[T]: # noqa: ARG002 

192 """Accept success handler (ignored) and continue short-circuit.""" 

193 return NeutralActionSuccess(result=self._result) 

194 

195 

196@final 

197class NeutralActionSuccess[T]: 

198 """Neutral action builder in success-configured state (no-op).""" 

199 

200 def __init__(self, *, result: Result[T] | None) -> None: 

201 """Initialize with failure result.""" 

202 self._result = result 

203 

204 def execute(self) -> ActionChain[T]: 

205 """Return failed chain without executing (short-circuit).""" 

206 return ActionChain(has_failed=True, result=self._result) 

207 

208 

209@final 

210class ActionChain[T]: 

211 """Executed action chain with result and railway track position. 

212 

213 Represents state after action execution: 

214 - Which rail? (success or failure) 

215 - What result? (Ok or Fail) 

216 

217 Provides .then() to chain additional actions with railway switching. 

218 

219 Example: 

220 >>> chain1 = action1.failure(h1).success(h2).execute() 

221 >>> if chain1.is_ok(): 

222 ... chain2 = chain1.then(action2).failure(h3).success(h4).execute() 

223 

224 """ 

225 

226 def __init__(self, *, has_failed: bool, result: Result[T] | None) -> None: 

227 """Initialize with execution result and failure status.""" 

228 self._has_failed = has_failed 

229 self._result = result 

230 

231 def result(self) -> Result[T] | None: 

232 """Get execution result (Ok, Fail, or None if skipped).""" 

233 return self._result 

234 

235 def then( 

236 self, action_or_start: Action[T] | ActionStart[T] 

237 ) -> ActionStart[T] | NeutralActionStart[T]: 

238 """Chain another action with railway switching logic. 

239 

240 Railway switching: 

241 - If succeeded: return ActionStart (action executes) 

242 - If failed: return NeutralActionStart (short-circuit) 

243 

244 Example: 

245 >>> chain2 = chain1.then(action2).failure(h).success(h).execute() 

246 >>> # action2 only executes if chain1 succeeded 

247 

248 """ 

249 if self._has_failed: 

250 return NeutralActionStart(result=self._result) 

251 

252 if isinstance(action_or_start, ActionStart): 

253 action = action_or_start.__action__ 

254 else: 

255 action = action_or_start 

256 

257 return ActionStart(action) 

258 

259 def has_failed(self) -> bool: 

260 """Check if chain is on failure rail.""" 

261 return self._has_failed 

262 

263 def is_ok(self) -> bool: 

264 """Check if chain is on success rail.""" 

265 return not self._has_failed