Coverage for async_durable_execution/_runner/local/processors/step.py: 99%
73 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-08-30 23:43 +0000
1"""Step operation processor for handling STEP operation updates."""
3from __future__ import annotations
5from datetime import datetime, timedelta, timezone
6from typing import Any
8from ...._core import (
9 ErrorObject,
10 Operation,
11 OperationAction,
12 OperationStatus,
13 OperationUpdate,
14 StepDetails,
15)
16from .base import (
17 OperationProcessor,
18)
19from ...exceptions import (
20 InvalidParameterValueException,
21)
22from ..time_scale import scale_delay
24VALID_ACTIONS_FOR_STEP = frozenset(
25 [
26 OperationAction.START,
27 OperationAction.FAIL,
28 OperationAction.RETRY,
29 OperationAction.SUCCEED,
30 ]
31)
34class StepProcessor(OperationProcessor):
35 """Processes STEP operation updates with retry scheduling."""
37 valid_actions = VALID_ACTIONS_FOR_STEP
38 _ALLOWED_STATUS_TO_CLOSE = frozenset(
39 [
40 OperationStatus.STARTED,
41 OperationStatus.READY,
42 ]
43 )
44 _ALLOWED_STATUS_TO_START = frozenset(
45 [
46 OperationStatus.READY,
47 ]
48 )
49 _ALLOWED_STATUS_TO_REATTEMPT = frozenset(
50 [
51 OperationStatus.STARTED,
52 OperationStatus.READY,
53 ]
54 )
56 @classmethod
57 def validate(cls, current_state: Operation | None, update: OperationUpdate) -> None:
58 """Validate STEP operation update."""
59 super().validate(current_state, update)
60 if current_state is None:
61 return
63 match update.action:
64 case OperationAction.START:
65 if current_state.status not in cls._ALLOWED_STATUS_TO_START:
66 msg_step_start: str = "Invalid current STEP state to start."
67 raise InvalidParameterValueException(msg_step_start)
68 case OperationAction.FAIL | OperationAction.SUCCEED:
69 if current_state.status not in cls._ALLOWED_STATUS_TO_CLOSE:
70 msg_step_close: str = "Invalid current STEP state to close."
71 raise InvalidParameterValueException(msg_step_close)
72 if update.action == OperationAction.FAIL and update.payload is not None:
73 msg_fail_payload: str = "Cannot provide a Payload for FAIL action."
74 raise InvalidParameterValueException(msg_fail_payload)
75 if (
76 update.action == OperationAction.SUCCEED
77 and update.error is not None
78 ):
79 msg_succeed_error: str = (
80 "Cannot provide an Error for SUCCEED action."
81 )
82 raise InvalidParameterValueException(msg_succeed_error)
83 case OperationAction.RETRY: 83 ↛ exitline 83 didn't return from function 'validate' because the pattern on line 83 always matched
84 if current_state.status not in cls._ALLOWED_STATUS_TO_REATTEMPT:
85 msg_step_retry: str = "Invalid current STEP state to re-attempt."
86 raise InvalidParameterValueException(msg_step_retry)
87 if update.step_options is None:
88 msg_step_options: str = "Invalid StepOptions for the given action."
89 raise InvalidParameterValueException(msg_step_options)
90 if update.error is not None and update.payload is not None:
91 msg_retry_both: str = (
92 "Cannot provide both error and payload to RETRY a STEP."
93 )
94 raise InvalidParameterValueException(msg_retry_both)
96 def process(
97 self,
98 update: OperationUpdate,
99 current_op: Operation | None,
100 notifier: Any,
101 execution_arn: str,
102 ) -> Operation:
103 """Process STEP operation update with scheduler integration for retries."""
104 match update.action:
105 case OperationAction.START:
106 return self._translate_update_to_operation(
107 update=update,
108 current_operation=current_op,
109 status=OperationStatus.STARTED,
110 )
111 case OperationAction.RETRY:
112 # set Status=PENDING, next attempt time, attempt count + 1
113 delay = (
114 update.step_options.next_attempt_delay_seconds
115 if update.step_options
116 else 0
117 )
118 scaled_delay = scale_delay(delay)
119 next_attempt_time = datetime.now(timezone.utc) + timedelta(
120 seconds=scaled_delay
121 )
123 # Build new step_details with incremented attempt
124 current_attempt = (
125 current_op.step_details.attempt
126 if current_op and current_op.step_details
127 else 0
128 )
129 previous_result = (
130 current_op.step_details.result
131 if current_op and current_op.step_details
132 else None
133 )
134 previous_error = (
135 current_op.step_details.error
136 if current_op and current_op.step_details
137 else None
138 )
139 result: str | None
140 error: ErrorObject | None
141 if update.payload is not None:
142 result = update.payload
143 error = update.error
144 elif update.error is not None:
145 result = previous_result
146 error = update.error
147 else:
148 result = previous_result
149 error = previous_error
151 new_step_details = StepDetails(
152 attempt=current_attempt + 1,
153 next_attempt_timestamp=next_attempt_time,
154 result=result,
155 error=error,
156 )
158 # Create new operation with updated step_details
159 retry_operation = Operation(
160 operation_id=update.operation_id,
161 operation_type=update.operation_type,
162 status=OperationStatus.PENDING,
163 parent_id=update.parent_id,
164 name=update.name,
165 start_timestamp=(
166 current_op.start_timestamp
167 if current_op
168 else datetime.now(timezone.utc)
169 ),
170 end_timestamp=None,
171 sub_type=update.sub_type,
172 execution_details=current_op.execution_details
173 if current_op
174 else None,
175 context_details=current_op.context_details if current_op else None,
176 step_details=new_step_details,
177 wait_details=current_op.wait_details if current_op else None,
178 callback_details=current_op.callback_details
179 if current_op
180 else None,
181 chained_invoke_details=current_op.chained_invoke_details
182 if current_op
183 else None,
184 )
186 # Schedule step retry timer to fire after delay
187 notifier.schedule_step_retry(
188 execution_arn, update.operation_id, scaled_delay
189 )
190 return retry_operation
191 case OperationAction.SUCCEED:
192 return self._translate_update_to_operation(
193 update=update,
194 current_operation=current_op,
195 status=OperationStatus.SUCCEEDED,
196 )
197 case OperationAction.FAIL:
198 return self._translate_update_to_operation(
199 update=update,
200 current_operation=current_op,
201 status=OperationStatus.FAILED,
202 )
203 case _:
204 msg: str = "Invalid action for STEP operation."
206 raise InvalidParameterValueException(msg)