Coverage for async-durable-execution/src/async_durable_execution/runner/local/processors/base.py: 100%
65 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-07 16:54 +0000
1"""Base processor class for operation transformations."""
3from __future__ import annotations
5import datetime
6from datetime import timedelta
7from typing import Any, ClassVar
9from ....models import (
10 CallbackDetails,
11 ChainedInvokeDetails,
12 ContextDetails,
13 ExecutionDetails,
14 Operation,
15 OperationAction,
16 OperationStatus,
17 OperationType,
18 OperationUpdate,
19 StepDetails,
20 WaitDetails,
21)
22from ...exceptions import (
23 InvalidParameterValueException,
24)
27class OperationProcessor:
28 """Base class for processing OperationUpdate to Operation transformations."""
30 valid_actions: ClassVar[frozenset[OperationAction]] = frozenset()
32 @classmethod
33 def validate(cls, current_state: Operation | None, update: OperationUpdate) -> None:
34 """Validate operation action and transition before processing."""
35 cls._validate_action(update.action)
37 @classmethod
38 def _validate_action(cls, action: OperationAction) -> None:
39 """Validate that an action is supported by this operation processor."""
40 if action not in cls.valid_actions:
41 msg = "Invalid action for the given operation type."
42 raise InvalidParameterValueException(msg)
44 def process(
45 self,
46 update: OperationUpdate,
47 current_op: Operation | None,
48 notifier: Any,
49 execution_arn: str,
50 ) -> Operation | None:
51 """Process an operation update and return the transformed operation."""
52 raise NotImplementedError
54 def _get_start_time(
55 self, current_operation: Operation | None
56 ) -> datetime.datetime | None:
57 start_time: datetime.datetime | None = (
58 current_operation.start_timestamp
59 if current_operation
60 else datetime.datetime.now(tz=datetime.timezone.utc)
61 )
62 return start_time
64 def _get_end_time(
65 self, current_operation: Operation | None, status: OperationStatus
66 ) -> datetime.datetime | None:
67 """Get end timestamp for operation based on current state and status."""
68 if current_operation and current_operation.end_timestamp:
69 return current_operation.end_timestamp
70 if status in {
71 OperationStatus.SUCCEEDED,
72 OperationStatus.FAILED,
73 OperationStatus.CANCELLED,
74 OperationStatus.TIMED_OUT,
75 OperationStatus.STOPPED,
76 }:
77 return datetime.datetime.now(tz=datetime.timezone.utc)
78 return None
80 def _create_execution_details(
81 self, update: OperationUpdate
82 ) -> ExecutionDetails | None:
83 """Create ExecutionDetails from OperationUpdate."""
84 return (
85 ExecutionDetails(input_payload=update.payload)
86 if update.operation_type == OperationType.EXECUTION
87 else None
88 )
90 def _create_context_details(self, update: OperationUpdate) -> ContextDetails | None:
91 """Create ContextDetails from OperationUpdate."""
92 return (
93 ContextDetails(
94 result=update.payload,
95 error=update.error,
96 replay_children=update.context_options.replay_children
97 if update.context_options
98 else False,
99 )
100 if update.operation_type == OperationType.CONTEXT
101 else None
102 )
104 def _create_step_details(
105 self,
106 update: OperationUpdate,
107 current_operation: Operation | None = None,
108 ) -> StepDetails | None:
109 """Create StepDetails from OperationUpdate.
111 Automatically increments attempt count for RETRY, SUCCEED, and FAIL actions.
112 """
114 attempt: int = 0
115 next_attempt_timestamp: datetime.datetime | None = None
117 if update.operation_type is OperationType.STEP:
118 if current_operation and current_operation.step_details:
119 attempt = current_operation.step_details.attempt
120 next_attempt_timestamp = (
121 current_operation.step_details.next_attempt_timestamp
122 )
123 # Increment attempt for RETRY, SUCCEED, and FAIL actions
124 if update.action in {
125 OperationAction.RETRY,
126 OperationAction.SUCCEED,
127 OperationAction.FAIL,
128 }:
129 attempt += 1
130 return StepDetails(
131 attempt=attempt,
132 next_attempt_timestamp=next_attempt_timestamp,
133 result=update.payload,
134 error=update.error,
135 )
137 return None
139 def _create_callback_details(
140 self, update: OperationUpdate
141 ) -> CallbackDetails | None:
142 """Create CallbackDetails from OperationUpdate."""
143 return (
144 CallbackDetails(
145 callback_id="placeholder", result=update.payload, error=update.error
146 )
147 if update.operation_type == OperationType.CALLBACK
148 else None
149 )
151 def _create_invoke_details(
152 self, update: OperationUpdate
153 ) -> ChainedInvokeDetails | None:
154 """Create ChainedInvokeDetails from OperationUpdate."""
155 if (
156 update.operation_type == OperationType.CHAINED_INVOKE
157 and update.chained_invoke_options
158 ):
159 return ChainedInvokeDetails(result=update.payload, error=update.error)
160 return None
162 def _translate_update_to_operation(
163 self,
164 update: OperationUpdate,
165 current_operation: Operation | None,
166 status: OperationStatus,
167 ) -> Operation:
168 """Transform OperationUpdate to Operation, always creating new Operation."""
169 start_time: datetime.datetime | None = self._get_start_time(current_operation)
170 end_time: datetime.datetime | None = self._get_end_time(
171 current_operation, status
172 )
174 execution_details = self._create_execution_details(update)
175 context_details = self._create_context_details(update)
176 step_details = self._create_step_details(update, current_operation)
177 callback_details = self._create_callback_details(update)
178 invoke_details = self._create_invoke_details(update)
179 wait_details = self._create_wait_details(update, current_operation)
181 return Operation(
182 operation_id=update.operation_id,
183 parent_id=update.parent_id,
184 name=update.name,
185 start_timestamp=start_time,
186 end_timestamp=end_time,
187 operation_type=update.operation_type,
188 status=status,
189 sub_type=update.sub_type,
190 execution_details=execution_details,
191 context_details=context_details,
192 step_details=step_details,
193 callback_details=callback_details,
194 chained_invoke_details=invoke_details,
195 wait_details=wait_details,
196 )
198 def _create_wait_details(
199 self, update: OperationUpdate, current_operation: Operation | None
200 ) -> WaitDetails | None:
201 """Create WaitDetails from OperationUpdate."""
202 if update.operation_type == OperationType.WAIT and update.wait_options:
203 if current_operation and current_operation.wait_details:
204 scheduled_end_timestamp = (
205 current_operation.wait_details.scheduled_end_timestamp
206 )
207 else:
208 scheduled_end_timestamp = datetime.datetime.now(
209 tz=datetime.timezone.utc
210 ) + timedelta(seconds=update.wait_options.wait_seconds)
211 return WaitDetails(scheduled_end_timestamp=scheduled_end_timestamp)
212 return None