Coverage for async-durable-execution/src/async_durable_execution/runner/model.py: 99%
1187 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-07 16:54 +0000
1"""Shared runner models."""
3from __future__ import annotations
5import datetime
6import json
7import logging
8from collections.abc import Mapping
9from dataclasses import dataclass, field, replace
10from enum import Enum
11from typing import Any, Protocol, TYPE_CHECKING
13from ..models import (
14 CallbackDetails,
15 CallbackOptions,
16 ChainedInvokeDetails,
17 ContextDetails,
18 DurableExecutionInvocationOutput,
19 ErrorObject,
20 ExecutionDetails,
21 InvocationStatus,
22 Operation,
23 OperationAction,
24 OperationPayload,
25 OperationStatus,
26 OperationSubType,
27 OperationType,
28 OperationUpdate,
29 StepDetails,
30 TimestampConverter,
31 WaitDetails,
32)
33from ..serdes import ExtendedTypeSerDes
34from .exceptions import (
35 DurableFunctionsTestError,
36 InvalidParameterValueException,
37)
39if TYPE_CHECKING:
40 from .local.model import StartDurableExecutionInput
43logger = logging.getLogger(__name__)
46class EventType(Enum):
47 """Event types for durable execution events."""
49 EXECUTION_STARTED = "ExecutionStarted"
50 EXECUTION_SUCCEEDED = "ExecutionSucceeded"
51 EXECUTION_FAILED = "ExecutionFailed"
52 EXECUTION_TIMED_OUT = "ExecutionTimedOut"
53 EXECUTION_STOPPED = "ExecutionStopped"
54 CONTEXT_STARTED = "ContextStarted"
55 CONTEXT_SUCCEEDED = "ContextSucceeded"
56 CONTEXT_FAILED = "ContextFailed"
57 WAIT_STARTED = "WaitStarted"
58 WAIT_SUCCEEDED = "WaitSucceeded"
59 WAIT_CANCELLED = "WaitCancelled"
60 STEP_STARTED = "StepStarted"
61 STEP_SUCCEEDED = "StepSucceeded"
62 STEP_FAILED = "StepFailed"
63 CHAINED_INVOKE_STARTED = "ChainedInvokeStarted"
64 CHAINED_INVOKE_SUCCEEDED = "ChainedInvokeSucceeded"
65 CHAINED_INVOKE_FAILED = "ChainedInvokeFailed"
66 CHAINED_INVOKE_TIMED_OUT = "ChainedInvokeTimedOut"
67 CHAINED_INVOKE_STOPPED = "ChainedInvokeStopped"
68 CALLBACK_STARTED = "CallbackStarted"
69 CALLBACK_SUCCEEDED = "CallbackSucceeded"
70 CALLBACK_FAILED = "CallbackFailed"
71 CALLBACK_TIMED_OUT = "CallbackTimedOut"
72 INVOCATION_COMPLETED = "InvocationCompleted"
75TERMINAL_STATUSES: set[OperationStatus] = {
76 OperationStatus.SUCCEEDED,
77 OperationStatus.FAILED,
78 OperationStatus.TIMED_OUT,
79 OperationStatus.STOPPED,
80 OperationStatus.CANCELLED,
81}
84@dataclass(frozen=True)
85class GetDurableExecutionResponse:
86 """Response containing durable execution details."""
88 durable_execution_arn: str
89 durable_execution_name: str
90 function_arn: str
91 status: str
92 start_timestamp: datetime.datetime
93 input_payload: str | None = None
94 result: str | None = None
95 error: ErrorObject | None = None
96 end_timestamp: datetime.datetime | None = None
97 version: str | None = None
99 @classmethod
100 def from_dict(cls, data: Mapping[str, Any]) -> GetDurableExecutionResponse:
101 error = None
102 if error_data := data.get("Error"):
103 error = ErrorObject.from_dict(error_data)
105 return cls(
106 durable_execution_arn=data["DurableExecutionArn"],
107 durable_execution_name=data["DurableExecutionName"],
108 function_arn=data["FunctionArn"],
109 status=data["Status"],
110 start_timestamp=data["StartTimestamp"],
111 input_payload=data.get("InputPayload"),
112 result=data.get("Result"),
113 error=error,
114 end_timestamp=data.get("EndTimestamp"),
115 version=data.get("Version"),
116 )
118 def to_dict(self) -> dict[str, Any]:
119 result: dict[str, Any] = {
120 "DurableExecutionArn": self.durable_execution_arn,
121 "DurableExecutionName": self.durable_execution_name,
122 "FunctionArn": self.function_arn,
123 "Status": self.status,
124 "StartTimestamp": self.start_timestamp,
125 }
126 if self.input_payload is not None:
127 result["InputPayload"] = self.input_payload
128 if self.result is not None:
129 result["Result"] = self.result
130 if self.error is not None:
131 result["Error"] = self.error.to_dict()
132 if self.end_timestamp is not None:
133 result["EndTimestamp"] = self.end_timestamp
134 if self.version is not None:
135 result["Version"] = self.version
136 return result
139# Event-related structures from Smithy model
140@dataclass(frozen=True)
141class EventInput:
142 """Event input structure."""
144 payload: str | None = None
145 truncated: bool = False
147 @classmethod
148 def from_dict(cls, data: dict) -> EventInput:
149 return cls(
150 payload=data.get("Payload"),
151 truncated=data.get("Truncated", False),
152 )
154 def to_dict(self) -> dict[str, Any]:
155 result: dict[str, Any] = {"Truncated": self.truncated}
156 if self.payload is not None:
157 result["Payload"] = self.payload
158 return result
160 @classmethod
161 def from_details(
162 cls,
163 details: ExecutionDetails,
164 include: bool = False, # noqa: FBT001, FBT002
165 ) -> EventInput:
166 details_input: str | None = details.input_payload if details else None
167 payload: str | None = details_input if include else None
168 truncated: bool = not include
169 return cls(payload=payload, truncated=truncated)
171 @classmethod
172 def from_start_durable_execution_input(
173 cls,
174 start_durable_execution_input: StartDurableExecutionInput,
175 include: bool = False, # noqa: FBT001, FBT002
176 ) -> EventInput:
177 input: str | None = start_durable_execution_input.input
178 truncated: bool = not include
179 return cls(input, truncated)
182@dataclass(frozen=True)
183class EventResult:
184 """Event result structure."""
186 payload: str | None = None
187 truncated: bool = False
189 @classmethod
190 def from_dict(cls, data: dict) -> EventResult:
191 return cls(
192 payload=data.get("Payload"),
193 truncated=data.get("Truncated", False),
194 )
196 def to_dict(self) -> dict[str, Any]:
197 result: dict[str, Any] = {"Truncated": self.truncated}
198 if self.payload is not None:
199 result["Payload"] = self.payload
200 return result
202 @classmethod
203 def from_details(
204 cls,
205 details: CallbackDetails | StepDetails | ChainedInvokeDetails | ContextDetails,
206 include: bool = False, # noqa: FBT001, FBT002
207 ) -> EventResult:
208 details_result: str | None = details.result if details else None
209 payload: str | None = details_result if include else None
210 truncated: bool = not include
211 return cls(payload=payload, truncated=truncated)
213 @classmethod
214 def from_durable_execution_invocation_output(
215 cls,
216 durable_execution_invocation_output: DurableExecutionInvocationOutput,
217 include: bool = False, # noqa: FBT001, FBT002
218 ) -> EventResult:
219 truncated: bool = not include
220 return cls(durable_execution_invocation_output.result, truncated)
223@dataclass(frozen=True)
224class EventError:
225 """Event error structure."""
227 payload: ErrorObject | None = None
228 truncated: bool = False
230 @classmethod
231 def from_dict(cls, data: dict) -> EventError:
232 payload = None
233 if payload_data := data.get("Payload"):
234 payload = ErrorObject.from_dict(payload_data)
236 return cls(
237 payload=payload,
238 truncated=data.get("Truncated", False),
239 )
241 def to_dict(self) -> dict[str, Any]:
242 result: dict[str, Any] = {"Truncated": self.truncated}
243 if self.payload is not None:
244 result["Payload"] = self.payload.to_dict()
245 return result
247 @classmethod
248 def from_details(
249 cls,
250 details: CallbackDetails | StepDetails | ChainedInvokeDetails | ContextDetails,
251 include: bool = False, # noqa: FBT001, FBT002
252 ) -> EventError:
253 error_object: ErrorObject | None = details.error if details else None
254 truncated: bool = not include
255 return cls(error_object, truncated)
257 @classmethod
258 def from_durable_execution_invocation_output(
259 cls,
260 durable_execution_invocation_output: DurableExecutionInvocationOutput,
261 include: bool = False, # noqa: FBT001, FBT002
262 ) -> EventError:
263 truncated: bool = not include
264 return cls(durable_execution_invocation_output.error, truncated)
267@dataclass(frozen=True)
268class RetryDetails:
269 """Retry details structure."""
271 current_attempt: int = 0
272 next_attempt_delay_seconds: int | None = None
274 @classmethod
275 def from_dict(cls, data: dict) -> RetryDetails:
276 return cls(
277 current_attempt=data.get("CurrentAttempt", 0),
278 next_attempt_delay_seconds=data.get("NextAttemptDelaySeconds"),
279 )
281 def to_dict(self) -> dict[str, Any]:
282 result: dict[str, Any] = {"CurrentAttempt": self.current_attempt}
283 if self.next_attempt_delay_seconds is not None:
284 result["NextAttemptDelaySeconds"] = self.next_attempt_delay_seconds
285 return result
288# Event detail structures
289@dataclass(frozen=True)
290class ExecutionStartedDetails:
291 """Execution started event details."""
293 input: EventInput | None = None
294 execution_timeout: int | None = None
296 @classmethod
297 def from_dict(cls, data: dict) -> ExecutionStartedDetails:
298 input_data = None
299 if input_dict := data.get("Input"):
300 input_data = EventInput.from_dict(input_dict)
302 return cls(
303 input=input_data,
304 execution_timeout=data.get("ExecutionTimeout"),
305 )
307 def to_dict(self) -> dict[str, Any]:
308 result: dict[str, Any] = {}
309 if self.input is not None:
310 result["Input"] = self.input.to_dict()
311 if self.execution_timeout is not None:
312 result["ExecutionTimeout"] = self.execution_timeout
313 return result
316@dataclass(frozen=True)
317class ExecutionSucceededDetails:
318 """Execution succeeded event details."""
320 result: EventResult | None = None
322 @classmethod
323 def from_dict(cls, data: dict) -> ExecutionSucceededDetails:
324 result_data = None
325 if result_dict := data.get("Result"):
326 result_data = EventResult.from_dict(result_dict)
328 return cls(result=result_data)
330 def to_dict(self) -> dict[str, Any]:
331 result: dict[str, Any] = {}
332 if self.result is not None:
333 result["Result"] = self.result.to_dict()
334 return result
337@dataclass(frozen=True)
338class ExecutionFailedDetails:
339 """Execution failed event details."""
341 error: EventError | None = None
343 @classmethod
344 def from_dict(cls, data: dict) -> ExecutionFailedDetails:
345 error_data = None
346 if error_dict := data.get("Error"):
347 error_data = EventError.from_dict(error_dict)
349 return cls(error=error_data)
351 def to_dict(self) -> dict[str, Any]:
352 result: dict[str, Any] = {}
353 if self.error is not None:
354 result["Error"] = self.error.to_dict()
355 return result
358@dataclass(frozen=True)
359class ExecutionTimedOutDetails:
360 """Execution timed out event details."""
362 error: EventError | None = None
364 @classmethod
365 def from_dict(cls, data: dict) -> ExecutionTimedOutDetails:
366 error_data = None
367 if error_dict := data.get("Error"):
368 error_data = EventError.from_dict(error_dict)
370 return cls(error=error_data)
372 def to_dict(self) -> dict[str, Any]:
373 result: dict[str, Any] = {}
374 if self.error is not None:
375 result["Error"] = self.error.to_dict()
376 return result
379@dataclass(frozen=True)
380class ExecutionStoppedDetails:
381 """Execution stopped event details."""
383 error: EventError | None = None
385 @classmethod
386 def from_dict(cls, data: dict) -> ExecutionStoppedDetails:
387 error_data = None
388 if error_dict := data.get("Error"):
389 error_data = EventError.from_dict(error_dict)
391 return cls(error=error_data)
393 def to_dict(self) -> dict[str, Any]:
394 result: dict[str, Any] = {}
395 if self.error is not None:
396 result["Error"] = self.error.to_dict()
397 return result
400@dataclass(frozen=True)
401class ContextStartedDetails:
402 """Context started event details."""
404 @classmethod
405 def from_dict(cls, data: dict) -> ContextStartedDetails: # noqa: ARG003
406 return cls()
408 def to_dict(self) -> dict[str, Any]:
409 return {}
412@dataclass(frozen=True)
413class ContextSucceededDetails:
414 """Context succeeded event details."""
416 result: EventResult | None = None
418 @classmethod
419 def from_dict(cls, data: dict) -> ContextSucceededDetails:
420 result_data = None
421 if result_dict := data.get("Result"):
422 result_data = EventResult.from_dict(result_dict)
424 return cls(result=result_data)
426 def to_dict(self) -> dict[str, Any]:
427 result: dict[str, Any] = {}
428 if self.result is not None:
429 result["Result"] = self.result.to_dict()
430 return result
433@dataclass(frozen=True)
434class ContextFailedDetails:
435 """Context failed event details."""
437 error: EventError | None = None
439 @classmethod
440 def from_dict(cls, data: dict) -> ContextFailedDetails:
441 error_data = None
442 if error_dict := data.get("Error"):
443 error_data = EventError.from_dict(error_dict)
445 return cls(error=error_data)
447 def to_dict(self) -> dict[str, Any]:
448 result: dict[str, Any] = {}
449 if self.error is not None:
450 result["Error"] = self.error.to_dict()
451 return result
454@dataclass(frozen=True)
455class WaitStartedDetails:
456 """Wait started event details."""
458 duration: int | None = None
459 scheduled_end_timestamp: datetime.datetime | None = None
461 @classmethod
462 def from_dict(cls, data: dict) -> WaitStartedDetails:
463 return cls(
464 duration=data.get("Duration"),
465 scheduled_end_timestamp=data.get("ScheduledEndTimestamp"),
466 )
468 def to_dict(self) -> dict[str, Any]:
469 result: dict[str, Any] = {}
470 if self.duration is not None:
471 result["Duration"] = self.duration
472 if self.scheduled_end_timestamp is not None:
473 result["ScheduledEndTimestamp"] = self.scheduled_end_timestamp
474 return result
477@dataclass(frozen=True)
478class WaitSucceededDetails:
479 """Wait succeeded event details."""
481 duration: int | None = None
483 @classmethod
484 def from_dict(cls, data: dict) -> WaitSucceededDetails:
485 return cls(duration=data.get("Duration"))
487 def to_dict(self) -> dict[str, Any]:
488 result: dict[str, Any] = {}
489 if self.duration is not None:
490 result["Duration"] = self.duration
491 return result
494@dataclass(frozen=True)
495class WaitCancelledDetails:
496 """Wait cancelled event details."""
498 error: EventError | None = None
500 @classmethod
501 def from_dict(cls, data: dict) -> WaitCancelledDetails:
502 error_data = None
503 if error_dict := data.get("Error"):
504 error_data = EventError.from_dict(error_dict)
506 return cls(error=error_data)
508 def to_dict(self) -> dict[str, Any]:
509 result: dict[str, Any] = {}
510 if self.error is not None:
511 result["Error"] = self.error.to_dict()
512 return result
515@dataclass(frozen=True)
516class StepStartedDetails:
517 """Step started event details."""
519 @classmethod
520 def from_dict(cls, data: dict) -> StepStartedDetails: # noqa: ARG003
521 return cls()
523 def to_dict(self) -> dict[str, Any]:
524 return {}
527@dataclass(frozen=True)
528class StepSucceededDetails:
529 """Step succeeded event details."""
531 result: EventResult | None = None
532 retry_details: RetryDetails | None = None
534 @classmethod
535 def from_dict(cls, data: dict) -> StepSucceededDetails:
536 result_data = None
537 if result_dict := data.get("Result"):
538 result_data = EventResult.from_dict(result_dict)
540 retry_details_data = None
541 if retry_dict := data.get("RetryDetails"):
542 retry_details_data = RetryDetails.from_dict(retry_dict)
544 return cls(result=result_data, retry_details=retry_details_data)
546 def to_dict(self) -> dict[str, Any]:
547 result: dict[str, Any] = {}
548 if self.result is not None:
549 result["Result"] = self.result.to_dict()
550 if self.retry_details is not None:
551 result["RetryDetails"] = self.retry_details.to_dict()
552 return result
555@dataclass(frozen=True)
556class StepFailedDetails:
557 """Step failed event details."""
559 error: EventError | None = None
560 retry_details: RetryDetails | None = None
562 @classmethod
563 def from_dict(cls, data: dict) -> StepFailedDetails:
564 error_data = None
565 if error_dict := data.get("Error"):
566 error_data = EventError.from_dict(error_dict)
568 retry_details_data = None
569 if retry_dict := data.get("RetryDetails"):
570 retry_details_data = RetryDetails.from_dict(retry_dict)
572 return cls(error=error_data, retry_details=retry_details_data)
574 def to_dict(self) -> dict[str, Any]:
575 result: dict[str, Any] = {}
576 if self.error is not None:
577 result["Error"] = self.error.to_dict()
578 if self.retry_details is not None:
579 result["RetryDetails"] = self.retry_details.to_dict()
580 return result
583@dataclass(frozen=True)
584class ChainedInvokePendingDetails:
585 """Chained Invoke Pending event details."""
587 input: EventInput | None = None
588 function_name: str | None = None
590 @classmethod
591 def from_dict(cls, data: dict) -> ChainedInvokePendingDetails:
592 input_data = None
593 if input_dict := data.get("Input"):
594 input_data = EventInput.from_dict(input_dict)
596 return cls(
597 input=input_data,
598 function_name=data.get("FunctionName"),
599 )
601 def to_dict(self) -> dict[str, Any]:
602 result: dict[str, Any] = {}
603 if self.input is not None:
604 result["Input"] = self.input.to_dict()
605 if self.function_name is not None:
606 result["FunctionName"] = self.function_name
607 return result
610@dataclass(frozen=True)
611class ChainedInvokeStartedDetails:
612 """Chained invoke started event details."""
614 durable_execution_arn: str | None = None
616 @classmethod
617 def from_dict(cls, data: dict) -> ChainedInvokeStartedDetails:
618 return cls(
619 durable_execution_arn=data.get("DurableExecutionArn"),
620 )
622 def to_dict(self) -> dict[str, Any]:
623 result: dict[str, Any] = {}
624 if self.durable_execution_arn is not None:
625 result["DurableExecutionArn"] = self.durable_execution_arn
626 return result
629@dataclass(frozen=True)
630class ChainedInvokeSucceededDetails:
631 """Chained invoke succeeded event details."""
633 result: EventResult | None = None
635 @classmethod
636 def from_dict(cls, data: dict) -> ChainedInvokeSucceededDetails:
637 result_data = None
638 if result_dict := data.get("Result"):
639 result_data = EventResult.from_dict(result_dict)
641 return cls(result=result_data)
643 def to_dict(self) -> dict[str, Any]:
644 result: dict[str, Any] = {}
645 if self.result is not None:
646 result["Result"] = self.result.to_dict()
647 return result
650@dataclass(frozen=True)
651class ChainedInvokeFailedDetails:
652 """Chained invoke failed event details."""
654 error: EventError | None = None
656 @classmethod
657 def from_dict(cls, data: dict) -> ChainedInvokeFailedDetails:
658 error_data = None
659 if error_dict := data.get("Error"):
660 error_data = EventError.from_dict(error_dict)
662 return cls(error=error_data)
664 def to_dict(self) -> dict[str, Any]:
665 result: dict[str, Any] = {}
666 if self.error is not None:
667 result["Error"] = self.error.to_dict()
668 return result
671@dataclass(frozen=True)
672class ChainedInvokeTimedOutDetails:
673 """Chained invoke timed out event details."""
675 error: EventError | None = None
677 @classmethod
678 def from_dict(cls, data: dict) -> ChainedInvokeTimedOutDetails:
679 error_data = None
680 if error_dict := data.get("Error"):
681 error_data = EventError.from_dict(error_dict)
683 return cls(error=error_data)
685 def to_dict(self) -> dict[str, Any]:
686 result: dict[str, Any] = {}
687 if self.error is not None:
688 result["Error"] = self.error.to_dict()
689 return result
692@dataclass(frozen=True)
693class ChainedInvokeStoppedDetails:
694 """Chained invoke stopped event details."""
696 error: EventError | None = None
698 @classmethod
699 def from_dict(cls, data: dict) -> ChainedInvokeStoppedDetails:
700 error_data = None
701 if error_dict := data.get("Error"):
702 error_data = EventError.from_dict(error_dict)
704 return cls(error=error_data)
706 def to_dict(self) -> dict[str, Any]:
707 result: dict[str, Any] = {}
708 if self.error is not None:
709 result["Error"] = self.error.to_dict()
710 return result
713@dataclass(frozen=True)
714class CallbackStartedDetails:
715 """Callback started event details."""
717 callback_id: str | None = None
718 heartbeat_timeout: int | None = None
719 timeout: int | None = None
721 @classmethod
722 def from_dict(cls, data: dict) -> CallbackStartedDetails:
723 return cls(
724 callback_id=data.get("CallbackId"),
725 heartbeat_timeout=data.get("HeartbeatTimeout"),
726 timeout=data.get("Timeout"),
727 )
729 def to_dict(self) -> dict[str, Any]:
730 result: dict[str, Any] = {}
731 if self.callback_id is not None:
732 result["CallbackId"] = self.callback_id
733 if self.heartbeat_timeout is not None:
734 result["HeartbeatTimeout"] = self.heartbeat_timeout
735 if self.timeout is not None:
736 result["Timeout"] = self.timeout
737 return result
740@dataclass(frozen=True)
741class CallbackSucceededDetails:
742 """Callback succeeded event details."""
744 result: EventResult | None = None
746 @classmethod
747 def from_dict(cls, data: dict) -> CallbackSucceededDetails:
748 result_data = None
749 if result_dict := data.get("Result"):
750 result_data = EventResult.from_dict(result_dict)
752 return cls(result=result_data)
754 def to_dict(self) -> dict[str, Any]:
755 result: dict[str, Any] = {}
756 if self.result is not None:
757 result["Result"] = self.result.to_dict()
758 return result
761@dataclass(frozen=True)
762class CallbackFailedDetails:
763 """Callback failed event details."""
765 error: EventError | None = None
767 @classmethod
768 def from_dict(cls, data: dict) -> CallbackFailedDetails:
769 error_data = None
770 if error_dict := data.get("Error"):
771 error_data = EventError.from_dict(error_dict)
773 return cls(error=error_data)
775 def to_dict(self) -> dict[str, Any]:
776 result: dict[str, Any] = {}
777 if self.error is not None:
778 result["Error"] = self.error.to_dict()
779 return result
782@dataclass(frozen=True)
783class CallbackTimedOutDetails:
784 """Callback timed out event details."""
786 error: EventError | None = None
788 @classmethod
789 def from_dict(cls, data: dict) -> CallbackTimedOutDetails:
790 error_data = None
791 if error_dict := data.get("Error"):
792 error_data = EventError.from_dict(error_dict)
794 return cls(error=error_data)
796 def to_dict(self) -> dict[str, Any]:
797 result: dict[str, Any] = {}
798 if self.error is not None:
799 result["Error"] = self.error.to_dict()
800 return result
803@dataclass(frozen=True)
804class InvocationCompletedDetails:
805 """Invocation completed event details."""
807 start_timestamp: datetime.datetime
808 end_timestamp: datetime.datetime
809 request_id: str
811 @classmethod
812 def from_dict(cls, data: dict) -> InvocationCompletedDetails:
813 return cls(
814 start_timestamp=data["StartTimestamp"],
815 end_timestamp=data["EndTimestamp"],
816 request_id=data["RequestId"],
817 )
819 @classmethod
820 def from_json_dict(cls, data: dict) -> InvocationCompletedDetails:
821 """Deserialize from JSON dict with Unix millisecond timestamps."""
822 start_ts: datetime.datetime | None = TimestampConverter.from_unix_millis(
823 data["StartTimestamp"]
824 )
825 end_ts: datetime.datetime | None = TimestampConverter.from_unix_millis(
826 data["EndTimestamp"]
827 )
829 if start_ts is None or end_ts is None:
830 raise InvalidParameterValueException(
831 "StartTimestamp and EndTimestamp cannot be null"
832 )
834 return cls(
835 start_timestamp=start_ts,
836 end_timestamp=end_ts,
837 request_id=data["RequestId"],
838 )
840 def to_dict(self) -> dict[str, Any]:
841 return {
842 "StartTimestamp": self.start_timestamp,
843 "EndTimestamp": self.end_timestamp,
844 "RequestId": self.request_id,
845 }
847 def to_json_dict(self) -> dict[str, Any]:
848 """Convert to JSON-serializable dict with Unix millisecond timestamps."""
849 return {
850 "StartTimestamp": TimestampConverter.to_unix_millis(self.start_timestamp),
851 "EndTimestamp": TimestampConverter.to_unix_millis(self.end_timestamp),
852 "RequestId": self.request_id,
853 }
856@dataclass(frozen=True)
857class EventCreationContext:
858 operation: Operation
859 event_id: int
860 durable_execution_arn: str
861 start_durable_execution_input: StartDurableExecutionInput
862 durable_execution_invocation_output: DurableExecutionInvocationOutput | None = None
863 operation_update: OperationUpdate | None = None
864 include_execution_data: bool = False
866 @classmethod
867 def create(
868 cls,
869 operation: Operation,
870 event_id: int,
871 durable_execution_arn: str,
872 start_input: StartDurableExecutionInput,
873 result: DurableExecutionInvocationOutput | None = None,
874 operation_update: OperationUpdate | None = None,
875 include_execution_data: bool = False, # noqa: FBT001, FBT002
876 ) -> EventCreationContext:
877 return cls(
878 operation=operation,
879 event_id=event_id,
880 durable_execution_arn=durable_execution_arn,
881 start_durable_execution_input=start_input,
882 durable_execution_invocation_output=result,
883 operation_update=operation_update,
884 include_execution_data=include_execution_data,
885 )
887 @property
888 def sub_type(self) -> str | None:
889 return self.operation.sub_type.value if self.operation.sub_type else None
891 def get_retry_details(self) -> RetryDetails | None:
892 if not self.operation.step_details or not self.operation_update:
893 return None
895 delay = 0
896 if (
897 self.operation_update.operation_type == OperationType.STEP
898 and self.operation_update.step_options
899 ):
900 delay = self.operation_update.step_options.next_attempt_delay_seconds
902 return RetryDetails(
903 current_attempt=self.operation.step_details.attempt,
904 next_attempt_delay_seconds=delay,
905 )
907 @property
908 def start_timestamp(self) -> datetime.datetime:
909 return (
910 self.operation.start_timestamp
911 if self.operation.start_timestamp is not None
912 else datetime.datetime.now(datetime.timezone.utc)
913 )
915 @property
916 def end_timestamp(self) -> datetime.datetime:
917 return (
918 self.operation.end_timestamp
919 if self.operation.end_timestamp is not None
920 else datetime.datetime.now(datetime.timezone.utc)
921 )
924@dataclass(frozen=True)
925class Event:
926 """Event structure from Smithy model."""
928 event_type: str
929 event_timestamp: datetime.datetime
930 sub_type: str | None = None
931 event_id: int = 1
932 operation_id: str | None = None
933 name: str | None = None
934 parent_id: str | None = None
935 execution_started_details: ExecutionStartedDetails | None = None
936 execution_succeeded_details: ExecutionSucceededDetails | None = None
937 execution_failed_details: ExecutionFailedDetails | None = None
938 execution_timed_out_details: ExecutionTimedOutDetails | None = None
939 execution_stopped_details: ExecutionStoppedDetails | None = None
940 context_started_details: ContextStartedDetails | None = None
941 context_succeeded_details: ContextSucceededDetails | None = None
942 context_failed_details: ContextFailedDetails | None = None
943 wait_started_details: WaitStartedDetails | None = None
944 wait_succeeded_details: WaitSucceededDetails | None = None
945 wait_cancelled_details: WaitCancelledDetails | None = None
946 step_started_details: StepStartedDetails | None = None
947 step_succeeded_details: StepSucceededDetails | None = None
948 step_failed_details: StepFailedDetails | None = None
949 chained_invoke_pending_details: ChainedInvokePendingDetails | None = None
950 chained_invoke_started_details: ChainedInvokeStartedDetails | None = None
951 chained_invoke_succeeded_details: ChainedInvokeSucceededDetails | None = None
952 chained_invoke_failed_details: ChainedInvokeFailedDetails | None = None
953 chained_invoke_timed_out_details: ChainedInvokeTimedOutDetails | None = None
954 chained_invoke_stopped_details: ChainedInvokeStoppedDetails | None = None
955 callback_started_details: CallbackStartedDetails | None = None
956 callback_succeeded_details: CallbackSucceededDetails | None = None
957 callback_failed_details: CallbackFailedDetails | None = None
958 callback_timed_out_details: CallbackTimedOutDetails | None = None
959 invocation_completed_details: InvocationCompletedDetails | None = None
961 @classmethod
962 def from_dict(cls, data: dict) -> Event:
963 # Parse all the detail structures
964 execution_started_details = None
965 if details_data := data.get("ExecutionStartedDetails"):
966 execution_started_details = ExecutionStartedDetails.from_dict(details_data)
968 execution_succeeded_details = None
969 if details_data := data.get("ExecutionSucceededDetails"):
970 execution_succeeded_details = ExecutionSucceededDetails.from_dict(
971 details_data
972 )
974 execution_failed_details = None
975 if details_data := data.get("ExecutionFailedDetails"):
976 execution_failed_details = ExecutionFailedDetails.from_dict(details_data)
978 execution_timed_out_details = None
979 if details_data := data.get("ExecutionTimedOutDetails"):
980 execution_timed_out_details = ExecutionTimedOutDetails.from_dict(
981 details_data
982 )
984 execution_stopped_details = None
985 if details_data := data.get("ExecutionStoppedDetails"):
986 execution_stopped_details = ExecutionStoppedDetails.from_dict(details_data)
988 context_started_details = None
989 if details_data := data.get("ContextStartedDetails"):
990 context_started_details = ContextStartedDetails.from_dict(details_data)
992 context_succeeded_details = None
993 if details_data := data.get("ContextSucceededDetails"):
994 context_succeeded_details = ContextSucceededDetails.from_dict(details_data)
996 context_failed_details = None
997 if details_data := data.get("ContextFailedDetails"):
998 context_failed_details = ContextFailedDetails.from_dict(details_data)
1000 wait_started_details = None
1001 if details_data := data.get("WaitStartedDetails"):
1002 wait_started_details = WaitStartedDetails.from_dict(details_data)
1004 wait_succeeded_details = None
1005 if details_data := data.get("WaitSucceededDetails"):
1006 wait_succeeded_details = WaitSucceededDetails.from_dict(details_data)
1008 wait_cancelled_details = None
1009 if details_data := data.get("WaitCancelledDetails"):
1010 wait_cancelled_details = WaitCancelledDetails.from_dict(details_data)
1012 step_started_details = None
1013 if details_data := data.get("StepStartedDetails"):
1014 step_started_details = StepStartedDetails.from_dict(details_data)
1016 step_succeeded_details = None
1017 if details_data := data.get("StepSucceededDetails"):
1018 step_succeeded_details = StepSucceededDetails.from_dict(details_data)
1020 step_failed_details = None
1021 if details_data := data.get("StepFailedDetails"):
1022 step_failed_details = StepFailedDetails.from_dict(details_data)
1024 chained_invoke_pending_details = None
1025 if details_data := data.get("ChainedInvokePendingDetails"):
1026 chained_invoke_pending_details = ChainedInvokePendingDetails.from_dict(
1027 details_data
1028 )
1030 chained_invoke_started_details = None
1031 if details_data := data.get("ChainedInvokeStartedDetails"):
1032 chained_invoke_started_details = ChainedInvokeStartedDetails.from_dict(
1033 details_data
1034 )
1036 chained_invoke_succeeded_details = None
1037 if details_data := data.get("ChainedInvokeSucceededDetails"):
1038 chained_invoke_succeeded_details = ChainedInvokeSucceededDetails.from_dict(
1039 details_data
1040 )
1042 chained_invoke_failed_details = None
1043 if details_data := data.get("ChainedInvokeFailedDetails"):
1044 chained_invoke_failed_details = ChainedInvokeFailedDetails.from_dict(
1045 details_data
1046 )
1048 chained_invoke_timed_out_details = None
1049 if details_data := data.get("ChainedInvokeTimedOutDetails"):
1050 chained_invoke_timed_out_details = ChainedInvokeTimedOutDetails.from_dict(
1051 details_data
1052 )
1054 chained_invoke_stopped_details = None
1055 if details_data := data.get("ChainedInvokeStoppedDetails"):
1056 chained_invoke_stopped_details = ChainedInvokeStoppedDetails.from_dict(
1057 details_data
1058 )
1060 callback_started_details = None
1061 if details_data := data.get("CallbackStartedDetails"):
1062 callback_started_details = CallbackStartedDetails.from_dict(details_data)
1064 callback_succeeded_details = None
1065 if details_data := data.get("CallbackSucceededDetails"):
1066 callback_succeeded_details = CallbackSucceededDetails.from_dict(
1067 details_data
1068 )
1070 callback_failed_details = None
1071 if details_data := data.get("CallbackFailedDetails"):
1072 callback_failed_details = CallbackFailedDetails.from_dict(details_data)
1074 callback_timed_out_details = None
1075 if details_data := data.get("CallbackTimedOutDetails"):
1076 callback_timed_out_details = CallbackTimedOutDetails.from_dict(details_data)
1078 invocation_completed_details = None
1079 if details_data := data.get("InvocationCompletedDetails"):
1080 invocation_completed_details = InvocationCompletedDetails.from_dict(
1081 details_data
1082 )
1084 return cls(
1085 event_type=data["EventType"],
1086 event_timestamp=data["EventTimestamp"],
1087 sub_type=data.get("SubType"),
1088 event_id=data.get("EventId", 1),
1089 operation_id=data.get("Id"),
1090 name=data.get("Name"),
1091 parent_id=data.get("ParentId"),
1092 execution_started_details=execution_started_details,
1093 execution_succeeded_details=execution_succeeded_details,
1094 execution_failed_details=execution_failed_details,
1095 execution_timed_out_details=execution_timed_out_details,
1096 execution_stopped_details=execution_stopped_details,
1097 context_started_details=context_started_details,
1098 context_succeeded_details=context_succeeded_details,
1099 context_failed_details=context_failed_details,
1100 wait_started_details=wait_started_details,
1101 wait_succeeded_details=wait_succeeded_details,
1102 wait_cancelled_details=wait_cancelled_details,
1103 step_started_details=step_started_details,
1104 step_succeeded_details=step_succeeded_details,
1105 step_failed_details=step_failed_details,
1106 chained_invoke_pending_details=chained_invoke_pending_details,
1107 chained_invoke_started_details=chained_invoke_started_details,
1108 chained_invoke_succeeded_details=chained_invoke_succeeded_details,
1109 chained_invoke_failed_details=chained_invoke_failed_details,
1110 chained_invoke_timed_out_details=chained_invoke_timed_out_details,
1111 chained_invoke_stopped_details=chained_invoke_stopped_details,
1112 callback_started_details=callback_started_details,
1113 callback_succeeded_details=callback_succeeded_details,
1114 callback_failed_details=callback_failed_details,
1115 callback_timed_out_details=callback_timed_out_details,
1116 invocation_completed_details=invocation_completed_details,
1117 )
1119 def to_dict(self) -> dict[str, Any]:
1120 result: dict[str, Any] = {
1121 "EventType": self.event_type,
1122 "EventTimestamp": self.event_timestamp,
1123 "EventId": self.event_id,
1124 }
1125 if self.sub_type is not None:
1126 result["SubType"] = self.sub_type
1127 if self.operation_id is not None:
1128 result["Id"] = self.operation_id
1129 if self.name is not None:
1130 result["Name"] = self.name
1131 if self.parent_id is not None:
1132 result["ParentId"] = self.parent_id
1133 if self.execution_started_details is not None:
1134 result["ExecutionStartedDetails"] = self.execution_started_details.to_dict()
1135 if self.execution_succeeded_details is not None:
1136 result["ExecutionSucceededDetails"] = (
1137 self.execution_succeeded_details.to_dict()
1138 )
1139 if self.execution_failed_details is not None:
1140 result["ExecutionFailedDetails"] = self.execution_failed_details.to_dict()
1141 if self.execution_timed_out_details is not None:
1142 result["ExecutionTimedOutDetails"] = (
1143 self.execution_timed_out_details.to_dict()
1144 )
1145 if self.execution_stopped_details is not None:
1146 result["ExecutionStoppedDetails"] = self.execution_stopped_details.to_dict()
1147 if self.context_started_details is not None:
1148 result["ContextStartedDetails"] = self.context_started_details.to_dict()
1149 if self.context_succeeded_details is not None:
1150 result["ContextSucceededDetails"] = self.context_succeeded_details.to_dict()
1151 if self.context_failed_details is not None:
1152 result["ContextFailedDetails"] = self.context_failed_details.to_dict()
1153 if self.wait_started_details is not None:
1154 result["WaitStartedDetails"] = self.wait_started_details.to_dict()
1155 if self.wait_succeeded_details is not None:
1156 result["WaitSucceededDetails"] = self.wait_succeeded_details.to_dict()
1157 if self.wait_cancelled_details is not None:
1158 result["WaitCancelledDetails"] = self.wait_cancelled_details.to_dict()
1159 if self.step_started_details is not None:
1160 result["StepStartedDetails"] = self.step_started_details.to_dict()
1161 if self.step_succeeded_details is not None:
1162 result["StepSucceededDetails"] = self.step_succeeded_details.to_dict()
1163 if self.step_failed_details is not None:
1164 result["StepFailedDetails"] = self.step_failed_details.to_dict()
1165 if self.chained_invoke_pending_details is not None:
1166 result["ChainedInvokePendingDetails"] = (
1167 self.chained_invoke_pending_details.to_dict()
1168 )
1169 if self.chained_invoke_started_details is not None:
1170 result["ChainedInvokeStartedDetails"] = (
1171 self.chained_invoke_started_details.to_dict()
1172 )
1173 if self.chained_invoke_succeeded_details is not None:
1174 result["ChainedInvokeSucceededDetails"] = (
1175 self.chained_invoke_succeeded_details.to_dict()
1176 )
1177 if self.chained_invoke_failed_details is not None:
1178 result["ChainedInvokeFailedDetails"] = (
1179 self.chained_invoke_failed_details.to_dict()
1180 )
1181 if self.chained_invoke_timed_out_details is not None:
1182 result["ChainedInvokeTimedOutDetails"] = (
1183 self.chained_invoke_timed_out_details.to_dict()
1184 )
1185 if self.chained_invoke_stopped_details is not None:
1186 result["ChainedInvokeStoppedDetails"] = (
1187 self.chained_invoke_stopped_details.to_dict()
1188 )
1189 if self.callback_started_details is not None:
1190 result["CallbackStartedDetails"] = self.callback_started_details.to_dict()
1191 if self.callback_succeeded_details is not None:
1192 result["CallbackSucceededDetails"] = (
1193 self.callback_succeeded_details.to_dict()
1194 )
1195 if self.callback_failed_details is not None:
1196 result["CallbackFailedDetails"] = self.callback_failed_details.to_dict()
1197 if self.callback_timed_out_details is not None:
1198 result["CallbackTimedOutDetails"] = (
1199 self.callback_timed_out_details.to_dict()
1200 )
1201 if self.invocation_completed_details is not None:
1202 result["InvocationCompletedDetails"] = (
1203 self.invocation_completed_details.to_dict()
1204 )
1205 return result
1207 @classmethod
1208 def create_execution_event_started(cls, context: EventCreationContext) -> Event:
1209 execution_details: ExecutionDetails | None = context.operation.execution_details
1210 event_input: EventInput | None = (
1211 EventInput.from_details(execution_details, context.include_execution_data)
1212 if execution_details
1213 else None
1214 )
1215 execution_timeout: int | None = (
1216 context.start_durable_execution_input.execution_timeout_seconds
1217 )
1219 return cls(
1220 event_type=EventType.EXECUTION_STARTED.value,
1221 event_timestamp=context.start_timestamp,
1222 sub_type=context.sub_type,
1223 event_id=context.event_id,
1224 operation_id=context.operation.operation_id,
1225 name=context.operation.name,
1226 parent_id=context.operation.parent_id,
1227 execution_started_details=ExecutionStartedDetails(
1228 input=event_input,
1229 execution_timeout=execution_timeout,
1230 ),
1231 )
1233 @classmethod
1234 def create_execution_event_succeeded(cls, context: EventCreationContext) -> Event:
1235 result: EventResult | None = (
1236 EventResult.from_durable_execution_invocation_output(
1237 context.durable_execution_invocation_output,
1238 context.include_execution_data,
1239 )
1240 if context.durable_execution_invocation_output
1241 else None
1242 )
1243 return cls(
1244 event_type=EventType.EXECUTION_SUCCEEDED.value,
1245 event_timestamp=context.end_timestamp,
1246 sub_type=context.sub_type,
1247 event_id=context.event_id,
1248 operation_id=context.operation.operation_id,
1249 name=context.operation.name,
1250 parent_id=context.operation.parent_id,
1251 execution_succeeded_details=ExecutionSucceededDetails(result=result),
1252 )
1254 @classmethod
1255 def create_execution_event_failed(cls, context: EventCreationContext) -> Event:
1256 error: EventError | None = (
1257 EventError.from_durable_execution_invocation_output(
1258 context.durable_execution_invocation_output,
1259 include=context.include_execution_data,
1260 )
1261 if context.durable_execution_invocation_output
1262 else None
1263 )
1264 return cls(
1265 event_type=EventType.EXECUTION_FAILED.value,
1266 event_timestamp=context.end_timestamp,
1267 sub_type=context.sub_type,
1268 event_id=context.event_id,
1269 operation_id=context.operation.operation_id,
1270 name=context.operation.name,
1271 parent_id=context.operation.parent_id,
1272 execution_failed_details=ExecutionFailedDetails(error=error),
1273 )
1275 @classmethod
1276 def create_execution_event_timed_out(cls, context: EventCreationContext) -> Event:
1277 error: EventError | None = (
1278 EventError.from_durable_execution_invocation_output(
1279 context.durable_execution_invocation_output,
1280 include=context.include_execution_data,
1281 )
1282 if context.durable_execution_invocation_output
1283 else None
1284 )
1285 return cls(
1286 event_type=EventType.EXECUTION_TIMED_OUT.value,
1287 event_timestamp=context.end_timestamp,
1288 sub_type=context.sub_type,
1289 event_id=context.event_id,
1290 operation_id=context.operation.operation_id,
1291 name=context.operation.name,
1292 parent_id=context.operation.parent_id,
1293 execution_timed_out_details=ExecutionTimedOutDetails(error=error),
1294 )
1296 @classmethod
1297 def create_execution_event_stopped(cls, context: EventCreationContext) -> Event:
1298 error: EventError | None = (
1299 EventError.from_durable_execution_invocation_output(
1300 context.durable_execution_invocation_output,
1301 include=context.include_execution_data,
1302 )
1303 if context.durable_execution_invocation_output
1304 else None
1305 )
1306 return cls(
1307 event_type=EventType.EXECUTION_STOPPED.value,
1308 event_timestamp=context.end_timestamp,
1309 sub_type=context.sub_type,
1310 event_id=context.event_id,
1311 operation_id=context.operation.operation_id,
1312 name=context.operation.name,
1313 parent_id=context.operation.parent_id,
1314 execution_stopped_details=ExecutionStoppedDetails(error=error),
1315 )
1317 @classmethod
1318 def create_execution_event(cls, context: EventCreationContext) -> Event:
1319 """Create execution event based on action."""
1320 match context.operation.status:
1321 case OperationStatus.STARTED:
1322 return cls.create_execution_event_started(context)
1323 case OperationStatus.SUCCEEDED:
1324 return cls.create_execution_event_succeeded(context)
1325 case OperationStatus.FAILED:
1326 return cls.create_execution_event_failed(context)
1327 case OperationStatus.TIMED_OUT:
1328 return cls.create_execution_event_timed_out(context)
1329 case OperationStatus.STOPPED:
1330 return cls.create_execution_event_stopped(context)
1331 case _:
1332 msg = f"Operation status {context.operation.status} is not valid for execution operations. Valid statuses are: STARTED, SUCCEEDED, FAILED, TIMED_OUT, STOPPED"
1333 raise InvalidParameterValueException(msg)
1335 @classmethod
1336 def create_context_event_started(cls, context: EventCreationContext) -> Event:
1337 return cls(
1338 event_type=EventType.CONTEXT_STARTED.value,
1339 event_timestamp=context.start_timestamp,
1340 sub_type=context.sub_type,
1341 event_id=context.event_id,
1342 operation_id=context.operation.operation_id,
1343 name=context.operation.name,
1344 parent_id=context.operation.parent_id,
1345 context_started_details=ContextStartedDetails(),
1346 )
1348 @classmethod
1349 def create_context_event_succeeded(cls, context: EventCreationContext) -> Event:
1350 context_details: ContextDetails | None = context.operation.context_details
1351 event_result: EventResult | None = (
1352 EventResult.from_details(context_details, context.include_execution_data)
1353 if context_details
1354 else None
1355 )
1356 return cls(
1357 event_type=EventType.CONTEXT_SUCCEEDED.value,
1358 event_timestamp=context.end_timestamp,
1359 sub_type=context.sub_type,
1360 event_id=context.event_id,
1361 operation_id=context.operation.operation_id,
1362 name=context.operation.name,
1363 parent_id=context.operation.parent_id,
1364 context_succeeded_details=ContextSucceededDetails(result=event_result),
1365 )
1367 @classmethod
1368 def create_context_event_failed(cls, context: EventCreationContext) -> Event:
1369 context_details: ContextDetails | None = context.operation.context_details
1370 event_error: EventError | None = (
1371 EventError.from_details(context_details) if context_details else None
1372 )
1373 return cls(
1374 event_type=EventType.CONTEXT_FAILED.value,
1375 event_timestamp=context.end_timestamp,
1376 sub_type=context.sub_type,
1377 event_id=context.event_id,
1378 operation_id=context.operation.operation_id,
1379 name=context.operation.name,
1380 parent_id=context.operation.parent_id,
1381 context_failed_details=ContextFailedDetails(error=event_error),
1382 )
1384 @classmethod
1385 def create_context_event(cls, context: EventCreationContext) -> Event:
1386 """Create context event based on action."""
1387 match context.operation.status:
1388 case OperationStatus.STARTED:
1389 return cls.create_context_event_started(context)
1390 case OperationStatus.SUCCEEDED:
1391 return cls.create_context_event_succeeded(context)
1392 case OperationStatus.FAILED:
1393 return cls.create_context_event_failed(context)
1394 case _:
1395 msg = (
1396 f"Operation status {context.operation.status} is not valid for context operations. "
1397 f"Valid statuses are: STARTED, SUCCEEDED, FAILED"
1398 )
1399 raise InvalidParameterValueException(msg)
1401 @classmethod
1402 def create_wait_event_started(cls, context: EventCreationContext) -> Event:
1403 wait_details: WaitDetails | None = context.operation.wait_details
1404 scheduled_end_timestamp: datetime.datetime | None = (
1405 wait_details.scheduled_end_timestamp if wait_details else None
1406 )
1407 duration: int | None = None
1408 if (
1409 wait_details
1410 and wait_details.scheduled_end_timestamp
1411 and context.operation.start_timestamp
1412 ):
1413 duration = round(
1414 (
1415 wait_details.scheduled_end_timestamp
1416 - context.operation.start_timestamp
1417 ).total_seconds()
1418 )
1419 return cls(
1420 event_type=EventType.WAIT_STARTED.value,
1421 event_timestamp=context.start_timestamp,
1422 sub_type=context.sub_type,
1423 event_id=context.event_id,
1424 operation_id=context.operation.operation_id,
1425 name=context.operation.name,
1426 parent_id=context.operation.parent_id,
1427 wait_started_details=WaitStartedDetails(
1428 duration=duration,
1429 scheduled_end_timestamp=scheduled_end_timestamp,
1430 ),
1431 )
1433 @classmethod
1434 def create_wait_event_succeeded(cls, context: EventCreationContext) -> Event:
1435 wait_details: WaitDetails | None = context.operation.wait_details
1436 duration: int | None = None
1437 if (
1438 wait_details
1439 and wait_details.scheduled_end_timestamp
1440 and context.operation.start_timestamp
1441 ):
1442 duration = round(
1443 (
1444 wait_details.scheduled_end_timestamp - context.start_timestamp
1445 ).total_seconds()
1446 )
1447 return cls(
1448 event_type=EventType.WAIT_SUCCEEDED.value,
1449 event_timestamp=context.end_timestamp,
1450 sub_type=context.sub_type,
1451 event_id=context.event_id,
1452 operation_id=context.operation.operation_id,
1453 name=context.operation.name,
1454 parent_id=context.operation.parent_id,
1455 wait_succeeded_details=WaitSucceededDetails(duration=duration),
1456 )
1458 @classmethod
1459 def create_wait_event_cancelled(cls, context: EventCreationContext) -> Event:
1460 error: EventError | None = None
1461 if (
1462 context.operation_update
1463 and context.operation_update.operation_type == OperationType.WAIT
1464 and context.operation_update.action == OperationAction.CANCEL
1465 ):
1466 error = EventError(
1467 context.operation_update.error, not context.include_execution_data
1468 )
1469 return cls(
1470 event_type=EventType.WAIT_CANCELLED.value,
1471 event_timestamp=context.end_timestamp,
1472 sub_type=context.sub_type,
1473 event_id=context.event_id,
1474 operation_id=context.operation.operation_id,
1475 name=context.operation.name,
1476 parent_id=context.operation.parent_id,
1477 wait_cancelled_details=WaitCancelledDetails(error=error),
1478 )
1480 @classmethod
1481 def create_wait_event(cls, context: EventCreationContext) -> Event:
1482 """Create wait event based on action."""
1483 match context.operation.status:
1484 case OperationStatus.STARTED:
1485 return cls.create_wait_event_started(context)
1486 case OperationStatus.SUCCEEDED:
1487 return cls.create_wait_event_succeeded(context)
1488 case OperationStatus.CANCELLED:
1489 return cls.create_wait_event_cancelled(context)
1490 case _:
1491 msg = (
1492 f"Operation status {context.operation.status} is not valid for wait operations. "
1493 f"Valid statuses are: STARTED, SUCCEEDED, CANCELLED"
1494 )
1495 raise InvalidParameterValueException(msg)
1497 @classmethod
1498 def create_step_event_started(cls, context: EventCreationContext) -> Event:
1499 return cls(
1500 event_type=EventType.STEP_STARTED.value,
1501 event_timestamp=context.start_timestamp,
1502 sub_type=context.sub_type,
1503 event_id=context.event_id,
1504 operation_id=context.operation.operation_id,
1505 name=context.operation.name,
1506 parent_id=context.operation.parent_id,
1507 step_started_details=StepStartedDetails(),
1508 )
1510 @classmethod
1511 def create_step_event_succeeded(cls, context: EventCreationContext) -> Event:
1512 step_details: StepDetails | None = context.operation.step_details
1513 event_result: EventResult | None = (
1514 EventResult.from_details(step_details, context.include_execution_data)
1515 if step_details
1516 else None
1517 )
1518 return cls(
1519 event_type=EventType.STEP_SUCCEEDED.value,
1520 event_timestamp=context.end_timestamp,
1521 sub_type=context.sub_type,
1522 event_id=context.event_id,
1523 operation_id=context.operation.operation_id,
1524 name=context.operation.name,
1525 parent_id=context.operation.parent_id,
1526 step_succeeded_details=StepSucceededDetails(
1527 result=event_result,
1528 retry_details=context.get_retry_details(),
1529 ),
1530 )
1532 @classmethod
1533 def create_step_event_failed(cls, context: EventCreationContext) -> Event:
1534 step_details: StepDetails | None = context.operation.step_details
1535 event_error: EventError | None = (
1536 EventError.from_details(
1537 step_details, include=context.include_execution_data
1538 )
1539 if step_details
1540 else None
1541 )
1542 return cls(
1543 event_type=EventType.STEP_FAILED.value,
1544 event_timestamp=context.end_timestamp,
1545 sub_type=context.sub_type,
1546 event_id=context.event_id,
1547 operation_id=context.operation.operation_id,
1548 name=context.operation.name,
1549 parent_id=context.operation.parent_id,
1550 step_failed_details=StepFailedDetails(
1551 error=event_error,
1552 retry_details=context.get_retry_details(),
1553 ),
1554 )
1556 @classmethod
1557 def create_step_event(cls, context: EventCreationContext) -> Event:
1558 """Create step event based on action."""
1559 match context.operation.status:
1560 case OperationStatus.STARTED:
1561 return cls.create_step_event_started(context)
1562 case OperationStatus.SUCCEEDED:
1563 return cls.create_step_event_succeeded(context)
1564 case OperationStatus.FAILED:
1565 return cls.create_step_event_failed(context)
1566 case _:
1567 msg = (
1568 f"Operation status {context.operation.status} is not valid for step operations. "
1569 f"Valid statuses are: STARTED, SUCCEEDED, FAILED"
1570 )
1571 raise InvalidParameterValueException(msg)
1573 @classmethod
1574 def create_chained_invoke_event_pending(
1575 cls, context: EventCreationContext
1576 ) -> Event:
1577 input: EventInput = EventInput.from_start_durable_execution_input(
1578 context.start_durable_execution_input, context.include_execution_data
1579 )
1580 return cls(
1581 event_type=EventType.CHAINED_INVOKE_STARTED.value,
1582 event_timestamp=context.start_timestamp,
1583 sub_type=context.sub_type,
1584 event_id=context.event_id,
1585 operation_id=context.operation.operation_id,
1586 name=context.operation.name,
1587 parent_id=context.operation.parent_id,
1588 chained_invoke_pending_details=ChainedInvokePendingDetails(
1589 input=input,
1590 function_name=context.start_durable_execution_input.function_name,
1591 ),
1592 )
1594 @classmethod
1595 def create_chained_invoke_event_started(
1596 cls, context: EventCreationContext
1597 ) -> Event:
1598 return cls(
1599 event_type=EventType.CHAINED_INVOKE_STARTED.value,
1600 event_timestamp=context.start_timestamp,
1601 sub_type=context.sub_type,
1602 event_id=context.event_id,
1603 operation_id=context.operation.operation_id,
1604 name=context.operation.name,
1605 parent_id=context.operation.parent_id,
1606 chained_invoke_started_details=ChainedInvokeStartedDetails(
1607 durable_execution_arn=context.durable_execution_arn
1608 ),
1609 )
1611 @classmethod
1612 def create_chained_invoke_event_succeeded(
1613 cls, context: EventCreationContext
1614 ) -> Event:
1615 chained_invoke_details: ChainedInvokeDetails | None = (
1616 context.operation.chained_invoke_details
1617 )
1618 event_result: EventResult | None = (
1619 EventResult.from_details(
1620 chained_invoke_details, context.include_execution_data
1621 )
1622 if chained_invoke_details
1623 else None
1624 )
1625 return cls(
1626 event_type=EventType.CHAINED_INVOKE_SUCCEEDED.value,
1627 event_timestamp=context.end_timestamp,
1628 sub_type=context.sub_type,
1629 event_id=context.event_id,
1630 operation_id=context.operation.operation_id,
1631 name=context.operation.name,
1632 parent_id=context.operation.parent_id,
1633 chained_invoke_succeeded_details=ChainedInvokeSucceededDetails(
1634 result=event_result
1635 ),
1636 )
1638 @classmethod
1639 def create_chained_invoke_event_failed(cls, context: EventCreationContext) -> Event:
1640 chained_invoke_details: ChainedInvokeDetails | None = (
1641 context.operation.chained_invoke_details
1642 )
1643 event_error: EventError | None = (
1644 EventError.from_details(
1645 chained_invoke_details, include=context.include_execution_data
1646 )
1647 if chained_invoke_details
1648 else None
1649 )
1650 return cls(
1651 event_type=EventType.CHAINED_INVOKE_FAILED.value,
1652 event_timestamp=context.end_timestamp,
1653 sub_type=context.sub_type,
1654 event_id=context.event_id,
1655 operation_id=context.operation.operation_id,
1656 name=context.operation.name,
1657 parent_id=context.operation.parent_id,
1658 chained_invoke_failed_details=ChainedInvokeFailedDetails(error=event_error),
1659 )
1661 @classmethod
1662 def create_chained_invoke_event_timed_out(
1663 cls, context: EventCreationContext
1664 ) -> Event:
1665 chained_invoke_details: ChainedInvokeDetails | None = (
1666 context.operation.chained_invoke_details
1667 )
1668 event_error: EventError | None = (
1669 EventError.from_details(
1670 chained_invoke_details, include=context.include_execution_data
1671 )
1672 if chained_invoke_details
1673 else None
1674 )
1675 return cls(
1676 event_type=EventType.CHAINED_INVOKE_TIMED_OUT.value,
1677 event_timestamp=context.end_timestamp,
1678 sub_type=context.sub_type,
1679 event_id=context.event_id,
1680 operation_id=context.operation.operation_id,
1681 name=context.operation.name,
1682 parent_id=context.operation.parent_id,
1683 chained_invoke_timed_out_details=ChainedInvokeTimedOutDetails(
1684 error=event_error
1685 ),
1686 )
1688 @classmethod
1689 def create_chained_invoke_event_stopped(
1690 cls, context: EventCreationContext
1691 ) -> Event:
1692 chained_invoke_details: ChainedInvokeDetails | None = (
1693 context.operation.chained_invoke_details
1694 )
1695 event_error: EventError | None = (
1696 EventError.from_details(
1697 chained_invoke_details, include=context.include_execution_data
1698 )
1699 if chained_invoke_details
1700 else None
1701 )
1702 return cls(
1703 event_type=EventType.CHAINED_INVOKE_STOPPED.value,
1704 event_timestamp=context.end_timestamp,
1705 sub_type=context.sub_type,
1706 event_id=context.event_id,
1707 operation_id=context.operation.operation_id,
1708 name=context.operation.name,
1709 parent_id=context.operation.parent_id,
1710 chained_invoke_stopped_details=ChainedInvokeStoppedDetails(
1711 error=event_error
1712 ),
1713 )
1715 @classmethod
1716 def create_chained_invoke_event(cls, context: EventCreationContext) -> Event:
1717 """Create chained invoke event based on action."""
1718 match context.operation.status:
1719 case OperationStatus.PENDING:
1720 return cls.create_chained_invoke_event_pending(context)
1721 case OperationStatus.STARTED:
1722 return cls.create_chained_invoke_event_started(context)
1723 case OperationStatus.SUCCEEDED:
1724 return cls.create_chained_invoke_event_succeeded(context)
1725 case OperationStatus.FAILED:
1726 return cls.create_chained_invoke_event_failed(context)
1727 case OperationStatus.TIMED_OUT:
1728 return cls.create_chained_invoke_event_timed_out(context)
1729 case OperationStatus.STOPPED:
1730 return cls.create_chained_invoke_event_stopped(context)
1731 case _:
1732 msg = (
1733 f"Operation status {context.operation.status} is not valid for chained invoke operations. Valid statuses are: "
1734 f"STARTED, SUCCEEDED, FAILED, TIMED_OUT, STOPPED"
1735 )
1736 raise InvalidParameterValueException(msg)
1738 @classmethod
1739 def create_callback_event_started(cls, context: EventCreationContext) -> Event:
1740 callback_details: CallbackDetails | None = context.operation.callback_details
1741 callback_id: str | None = (
1742 callback_details.callback_id if callback_details else None
1743 )
1744 callback_options: CallbackOptions | None = (
1745 context.operation_update.callback_options
1746 if context.operation_update
1747 else None
1748 )
1749 timeout: int | None = (
1750 callback_options.timeout_seconds if callback_options else None
1751 )
1752 heartbeat_timeout: int | None = (
1753 callback_options.heartbeat_timeout_seconds if callback_options else None
1754 )
1755 return cls(
1756 event_type=EventType.CALLBACK_STARTED.value,
1757 event_timestamp=context.start_timestamp,
1758 sub_type=context.sub_type,
1759 event_id=context.event_id,
1760 operation_id=context.operation.operation_id,
1761 name=context.operation.name,
1762 parent_id=context.operation.parent_id,
1763 callback_started_details=CallbackStartedDetails(
1764 callback_id=callback_id,
1765 timeout=timeout,
1766 heartbeat_timeout=heartbeat_timeout,
1767 ),
1768 )
1770 @classmethod
1771 def create_callback_event_succeeded(cls, context: EventCreationContext) -> Event:
1772 callback_details: CallbackDetails | None = context.operation.callback_details
1773 event_result: EventResult | None = (
1774 EventResult.from_details(callback_details, context.include_execution_data)
1775 if callback_details
1776 else None
1777 )
1778 return cls(
1779 event_type=EventType.CALLBACK_SUCCEEDED.value,
1780 event_timestamp=context.end_timestamp,
1781 sub_type=context.sub_type,
1782 event_id=context.event_id,
1783 operation_id=context.operation.operation_id,
1784 name=context.operation.name,
1785 parent_id=context.operation.parent_id,
1786 callback_succeeded_details=CallbackSucceededDetails(result=event_result),
1787 )
1789 @classmethod
1790 def create_callback_event_failed(cls, context: EventCreationContext) -> Event:
1791 callback_details: CallbackDetails | None = context.operation.callback_details
1792 event_error: EventError | None = (
1793 EventError.from_details(callback_details) if callback_details else None
1794 )
1795 return cls(
1796 event_type=EventType.CALLBACK_FAILED.value,
1797 event_timestamp=context.end_timestamp,
1798 sub_type=context.sub_type,
1799 event_id=context.event_id,
1800 operation_id=context.operation.operation_id,
1801 name=context.operation.name,
1802 parent_id=context.operation.parent_id,
1803 callback_failed_details=CallbackFailedDetails(error=event_error),
1804 )
1806 @classmethod
1807 def create_callback_event_timed_out(cls, context: EventCreationContext) -> Event:
1808 callback_details: CallbackDetails | None = context.operation.callback_details
1809 event_error: EventError | None = (
1810 EventError.from_details(callback_details) if callback_details else None
1811 )
1812 return cls(
1813 event_type=EventType.CALLBACK_TIMED_OUT.value,
1814 event_timestamp=context.end_timestamp,
1815 sub_type=context.sub_type,
1816 event_id=context.event_id,
1817 operation_id=context.operation.operation_id,
1818 name=context.operation.name,
1819 parent_id=context.operation.parent_id,
1820 callback_timed_out_details=CallbackTimedOutDetails(error=event_error),
1821 )
1823 @classmethod
1824 def create_callback_event(cls, context: EventCreationContext) -> Event:
1825 """Create callback event based on action."""
1826 match context.operation.status:
1827 case OperationStatus.STARTED:
1828 return cls.create_callback_event_started(context)
1829 case OperationStatus.SUCCEEDED:
1830 return cls.create_callback_event_succeeded(context)
1831 case OperationStatus.FAILED:
1832 return cls.create_callback_event_failed(context)
1833 case OperationStatus.TIMED_OUT:
1834 return cls.create_callback_event_timed_out(context)
1835 case _:
1836 msg = (
1837 f"Operation status {context.operation.status} is not valid for callback operations. "
1838 f"Valid statuses are: STARTED, SUCCEEDED, FAILED, TIMED_OUT"
1839 )
1840 raise InvalidParameterValueException(msg)
1842 @classmethod
1843 def create_invocation_completed(
1844 cls,
1845 event_id: int,
1846 event_timestamp: datetime.datetime,
1847 start_timestamp: datetime.datetime,
1848 end_timestamp: datetime.datetime,
1849 request_id: str,
1850 ) -> Event:
1851 """Create invocation completed event."""
1852 return cls(
1853 event_type=EventType.INVOCATION_COMPLETED.value,
1854 event_timestamp=event_timestamp,
1855 event_id=event_id,
1856 invocation_completed_details=InvocationCompletedDetails(
1857 start_timestamp=start_timestamp,
1858 end_timestamp=end_timestamp,
1859 request_id=request_id,
1860 ),
1861 )
1863 @classmethod
1864 def create_event_started(cls, context: EventCreationContext) -> Event:
1865 """Convert operation to started event."""
1866 if context.operation.start_timestamp is None:
1867 msg: str = "Operation start timestamp cannot be None when converting to started event"
1868 raise InvalidParameterValueException(msg)
1870 match context.operation.operation_type:
1871 case OperationType.EXECUTION:
1872 return cls.create_execution_event_started(context)
1873 case OperationType.CONTEXT:
1874 return cls.create_context_event_started(context)
1875 case OperationType.WAIT:
1876 return cls.create_wait_event_started(context)
1877 case OperationType.STEP:
1878 return cls.create_step_event_started(context)
1879 case OperationType.CHAINED_INVOKE:
1880 return cls.create_chained_invoke_event_started(context)
1881 case OperationType.CALLBACK:
1882 return cls.create_callback_event_started(context)
1883 case _:
1884 msg = f"Unknown operation type: {context.operation.operation_type}"
1885 raise InvalidParameterValueException(msg)
1887 @classmethod
1888 def from_event_with_id(cls, event: Event, event_id: int) -> Event:
1889 """Create a new Event from an existing event with updated event_id."""
1890 return cls(
1891 event_type=event.event_type,
1892 event_timestamp=event.event_timestamp,
1893 sub_type=event.sub_type,
1894 event_id=event_id,
1895 operation_id=event.operation_id,
1896 name=event.name,
1897 parent_id=event.parent_id,
1898 execution_started_details=event.execution_started_details,
1899 execution_succeeded_details=event.execution_succeeded_details,
1900 execution_failed_details=event.execution_failed_details,
1901 execution_timed_out_details=event.execution_timed_out_details,
1902 execution_stopped_details=event.execution_stopped_details,
1903 context_started_details=event.context_started_details,
1904 context_succeeded_details=event.context_succeeded_details,
1905 context_failed_details=event.context_failed_details,
1906 wait_started_details=event.wait_started_details,
1907 wait_succeeded_details=event.wait_succeeded_details,
1908 wait_cancelled_details=event.wait_cancelled_details,
1909 step_started_details=event.step_started_details,
1910 step_succeeded_details=event.step_succeeded_details,
1911 step_failed_details=event.step_failed_details,
1912 chained_invoke_pending_details=event.chained_invoke_pending_details,
1913 chained_invoke_started_details=event.chained_invoke_started_details,
1914 chained_invoke_succeeded_details=event.chained_invoke_succeeded_details,
1915 chained_invoke_failed_details=event.chained_invoke_failed_details,
1916 chained_invoke_timed_out_details=event.chained_invoke_timed_out_details,
1917 chained_invoke_stopped_details=event.chained_invoke_stopped_details,
1918 callback_started_details=event.callback_started_details,
1919 callback_succeeded_details=event.callback_succeeded_details,
1920 callback_failed_details=event.callback_failed_details,
1921 callback_timed_out_details=event.callback_timed_out_details,
1922 )
1924 @classmethod
1925 def create_event_terminated(cls, context: EventCreationContext) -> Event:
1926 """Convert operation to finished event."""
1927 operation: Operation = context.operation
1928 if operation.end_timestamp is None:
1929 msg: str = "Operation end timestamp cannot be None when converting to finished event"
1930 raise InvalidParameterValueException(msg)
1932 if operation.status not in TERMINAL_STATUSES:
1933 msg = f"Operation status must be one of SUCCEEDED, FAILED, TIMED_OUT, STOPPED, or CANCELLED. Got: {operation.status}"
1934 raise InvalidParameterValueException(msg)
1936 match operation.operation_type:
1937 case OperationType.EXECUTION:
1938 return cls.create_execution_event(context)
1939 case OperationType.CONTEXT:
1940 return cls.create_context_event(context)
1941 case OperationType.WAIT:
1942 return cls.create_wait_event(context)
1943 case OperationType.STEP:
1944 return cls.create_step_event(context)
1945 case OperationType.CHAINED_INVOKE:
1946 return cls.create_chained_invoke_event(context)
1947 case OperationType.CALLBACK:
1948 return cls.create_callback_event(context)
1949 case _:
1950 msg = f"Unknown operation type: {operation.operation_type}"
1951 raise InvalidParameterValueException(msg)
1954@dataclass(frozen=True)
1955class HistoryEventTypeConfig:
1956 """Configuration for how to process a specific event type."""
1958 operation_type: OperationType | None
1959 operation_status: OperationStatus | None
1960 is_start_event: bool
1961 is_end_event: bool
1962 has_result: bool # Whether this event type contains result/error data
1965# Mapping of event types to their processing configuration
1966# This matches the TypeScript historyEventTypes constant
1967HISTORY_EVENT_TYPES: dict[str, HistoryEventTypeConfig] = {
1968 "ExecutionStarted": HistoryEventTypeConfig(
1969 operation_type=OperationType.EXECUTION,
1970 operation_status=OperationStatus.STARTED,
1971 is_start_event=True,
1972 is_end_event=False,
1973 has_result=False,
1974 ),
1975 "ExecutionFailed": HistoryEventTypeConfig(
1976 operation_type=OperationType.EXECUTION,
1977 operation_status=OperationStatus.FAILED,
1978 is_start_event=False,
1979 is_end_event=True,
1980 has_result=False,
1981 ),
1982 "ExecutionStopped": HistoryEventTypeConfig(
1983 operation_type=OperationType.EXECUTION,
1984 operation_status=OperationStatus.STOPPED,
1985 is_start_event=False,
1986 is_end_event=True,
1987 has_result=False,
1988 ),
1989 "ExecutionSucceeded": HistoryEventTypeConfig(
1990 operation_type=OperationType.EXECUTION,
1991 operation_status=OperationStatus.SUCCEEDED,
1992 is_start_event=False,
1993 is_end_event=True,
1994 has_result=False,
1995 ),
1996 "ExecutionTimedOut": HistoryEventTypeConfig(
1997 operation_type=OperationType.EXECUTION,
1998 operation_status=OperationStatus.TIMED_OUT,
1999 is_start_event=False,
2000 is_end_event=True,
2001 has_result=False,
2002 ),
2003 "CallbackStarted": HistoryEventTypeConfig(
2004 operation_type=OperationType.CALLBACK,
2005 operation_status=OperationStatus.STARTED,
2006 is_start_event=True,
2007 is_end_event=False,
2008 has_result=False,
2009 ),
2010 "CallbackFailed": HistoryEventTypeConfig(
2011 operation_type=OperationType.CALLBACK,
2012 operation_status=OperationStatus.FAILED,
2013 is_start_event=False,
2014 is_end_event=True,
2015 has_result=True,
2016 ),
2017 "CallbackSucceeded": HistoryEventTypeConfig(
2018 operation_type=OperationType.CALLBACK,
2019 operation_status=OperationStatus.SUCCEEDED,
2020 is_start_event=False,
2021 is_end_event=True,
2022 has_result=True,
2023 ),
2024 "CallbackTimedOut": HistoryEventTypeConfig(
2025 operation_type=OperationType.CALLBACK,
2026 operation_status=OperationStatus.TIMED_OUT,
2027 is_start_event=False,
2028 is_end_event=True,
2029 has_result=True,
2030 ),
2031 "ContextStarted": HistoryEventTypeConfig(
2032 operation_type=OperationType.CONTEXT,
2033 operation_status=OperationStatus.STARTED,
2034 is_start_event=True,
2035 is_end_event=False,
2036 has_result=False,
2037 ),
2038 "ContextFailed": HistoryEventTypeConfig(
2039 operation_type=OperationType.CONTEXT,
2040 operation_status=OperationStatus.FAILED,
2041 is_start_event=False,
2042 is_end_event=True,
2043 has_result=True,
2044 ),
2045 "ContextSucceeded": HistoryEventTypeConfig(
2046 operation_type=OperationType.CONTEXT,
2047 operation_status=OperationStatus.SUCCEEDED,
2048 is_start_event=False,
2049 is_end_event=True,
2050 has_result=True,
2051 ),
2052 "ChainedInvokeStarted": HistoryEventTypeConfig(
2053 operation_type=OperationType.CHAINED_INVOKE,
2054 operation_status=OperationStatus.STARTED,
2055 is_start_event=True,
2056 is_end_event=False,
2057 has_result=False,
2058 ),
2059 "ChainedInvokeFailed": HistoryEventTypeConfig(
2060 operation_type=OperationType.CHAINED_INVOKE,
2061 operation_status=OperationStatus.FAILED,
2062 is_start_event=False,
2063 is_end_event=True,
2064 has_result=True,
2065 ),
2066 "ChainedInvokeSucceeded": HistoryEventTypeConfig(
2067 operation_type=OperationType.CHAINED_INVOKE,
2068 operation_status=OperationStatus.SUCCEEDED,
2069 is_start_event=False,
2070 is_end_event=True,
2071 has_result=True,
2072 ),
2073 "ChainedInvokeTimedOut": HistoryEventTypeConfig(
2074 operation_type=OperationType.CHAINED_INVOKE,
2075 operation_status=OperationStatus.TIMED_OUT,
2076 is_start_event=False,
2077 is_end_event=True,
2078 has_result=True,
2079 ),
2080 "ChainedInvokeCancelled": HistoryEventTypeConfig(
2081 operation_type=OperationType.CHAINED_INVOKE,
2082 operation_status=OperationStatus.CANCELLED,
2083 is_start_event=False,
2084 is_end_event=True,
2085 has_result=True,
2086 ),
2087 "StepStarted": HistoryEventTypeConfig(
2088 operation_type=OperationType.STEP,
2089 operation_status=OperationStatus.STARTED,
2090 is_start_event=True,
2091 is_end_event=False,
2092 has_result=False,
2093 ),
2094 "StepFailed": HistoryEventTypeConfig(
2095 operation_type=OperationType.STEP,
2096 operation_status=OperationStatus.FAILED,
2097 is_start_event=False,
2098 is_end_event=True,
2099 has_result=True,
2100 ),
2101 "StepSucceeded": HistoryEventTypeConfig(
2102 operation_type=OperationType.STEP,
2103 operation_status=OperationStatus.SUCCEEDED,
2104 is_start_event=False,
2105 is_end_event=True,
2106 has_result=True,
2107 ),
2108 "WaitStarted": HistoryEventTypeConfig(
2109 operation_type=OperationType.WAIT,
2110 operation_status=OperationStatus.STARTED,
2111 is_start_event=True,
2112 is_end_event=False,
2113 has_result=True,
2114 ),
2115 "WaitSucceeded": HistoryEventTypeConfig(
2116 operation_type=OperationType.WAIT,
2117 operation_status=OperationStatus.SUCCEEDED,
2118 is_start_event=False,
2119 is_end_event=True,
2120 has_result=True,
2121 ),
2122 "WaitCancelled": HistoryEventTypeConfig(
2123 operation_type=OperationType.WAIT,
2124 operation_status=OperationStatus.CANCELLED,
2125 is_start_event=False,
2126 is_end_event=True,
2127 has_result=True,
2128 ),
2129 # TODO: add support for populating invocation information from InvocationCompleted event
2130 "InvocationCompleted": HistoryEventTypeConfig(
2131 operation_type=None,
2132 operation_status=None,
2133 is_start_event=False,
2134 is_end_event=False,
2135 has_result=True,
2136 ),
2137}
2140def events_to_operations(events: list[Event]) -> list[Operation]:
2141 """Convert a list of history events into operations.
2143 This function processes raw history events and groups them by operation ID,
2144 creating comprehensive operation objects following the TypeScript pattern from
2145 aws-durable-execution-sdk-js-testing.
2147 Multiple events for the same operation_id are merged together, with each event
2148 contributing its specific fields (e.g., CallbackStarted provides callback_id,
2149 CallbackSucceeded provides result).
2151 Args:
2152 events: List of history events to process
2154 Returns:
2155 List of operations, one per unique operation ID
2157 Raises:
2158 InvalidParameterValueException: When required fields are missing from an event
2160 Note:
2161 InvocationCompleted events are currently skipped as they don't represent
2162 operations. Future enhancement: populate invocation information from these
2163 events (TODO).
2164 """
2165 operations_map: dict[str, Operation] = {}
2167 for event in events:
2168 if not event.event_type:
2169 msg = "Missing required 'event_type' field in event"
2170 raise InvalidParameterValueException(msg)
2172 # Get event type configuration
2173 event_config: HistoryEventTypeConfig | None = HISTORY_EVENT_TYPES.get(
2174 event.event_type
2175 )
2176 if not event_config:
2177 msg = f"Unknown event type: {event.event_type}"
2178 raise InvalidParameterValueException(msg)
2180 # TODO: add support for populating invocation information from InvocationCompleted event
2181 if event.event_type == "InvocationCompleted":
2182 continue
2184 if not event.operation_id:
2185 msg = f"Missing required 'operation_id' field in event {event.event_id}"
2186 raise InvalidParameterValueException(msg)
2188 # Get previous operation if it exists
2189 previous_operation: Operation | None = operations_map.get(event.operation_id)
2191 # Get operation type and status from configuration
2192 operation_type: OperationType = (
2193 event_config.operation_type or OperationType.EXECUTION
2194 )
2195 status: OperationStatus = (
2196 event_config.operation_status or OperationStatus.PENDING
2197 )
2199 # Parse sub_type
2200 sub_type: OperationSubType | None = None
2201 if event.sub_type:
2202 try:
2203 sub_type = OperationSubType(event.sub_type)
2204 except ValueError as e:
2205 raise InvalidParameterValueException(str(e)) from e
2207 # Create base operation
2208 operation = Operation(
2209 operation_id=event.operation_id,
2210 operation_type=operation_type,
2211 status=status,
2212 name=event.name,
2213 parent_id=event.parent_id,
2214 sub_type=sub_type,
2215 start_timestamp=datetime.datetime.now(tz=datetime.timezone.utc),
2216 )
2218 # Merge with previous operation if it exists
2219 # Most fields are immutable, so they get preserved from previous events
2220 if previous_operation:
2221 operation = replace(
2222 operation,
2223 name=operation.name or previous_operation.name,
2224 parent_id=operation.parent_id or previous_operation.parent_id,
2225 sub_type=operation.sub_type or previous_operation.sub_type,
2226 start_timestamp=previous_operation.start_timestamp,
2227 end_timestamp=previous_operation.end_timestamp,
2228 execution_details=previous_operation.execution_details,
2229 context_details=previous_operation.context_details,
2230 step_details=previous_operation.step_details,
2231 wait_details=previous_operation.wait_details,
2232 callback_details=previous_operation.callback_details,
2233 chained_invoke_details=previous_operation.chained_invoke_details,
2234 )
2236 # Set timestamps based on event configuration
2237 if event_config.is_start_event:
2238 operation = replace(operation, start_timestamp=event.event_timestamp)
2239 if event_config.is_end_event:
2240 operation = replace(operation, end_timestamp=event.event_timestamp)
2242 # Add operation-specific details incrementally
2243 # Each event type contributes only the fields it has
2245 # EXECUTION details
2246 if (
2247 operation_type == OperationType.EXECUTION
2248 and event.execution_started_details
2249 and event.execution_started_details.input
2250 ):
2251 operation = replace(
2252 operation,
2253 execution_details=ExecutionDetails(
2254 input_payload=event.execution_started_details.input.payload
2255 ),
2256 )
2258 # CALLBACK details - merge callback_id, result, and error from different events
2259 if operation_type == OperationType.CALLBACK:
2260 existing_cb: CallbackDetails | None = operation.callback_details
2261 callback_id: str = existing_cb.callback_id if existing_cb else ""
2262 result: str | None = existing_cb.result if existing_cb else None
2263 error: ErrorObject | None = existing_cb.error if existing_cb else None
2265 # CallbackStarted provides callback_id
2266 if event.callback_started_details:
2267 callback_id = event.callback_started_details.callback_id or callback_id
2269 # CallbackSucceeded provides result
2270 if (
2271 event.callback_succeeded_details
2272 and event.callback_succeeded_details.result
2273 ):
2274 result = event.callback_succeeded_details.result.payload
2276 # CallbackFailed provides error
2277 if event.callback_failed_details and event.callback_failed_details.error:
2278 error = event.callback_failed_details.error.payload
2280 # CallbackTimedOut provides error
2281 if (
2282 event.callback_timed_out_details
2283 and event.callback_timed_out_details.error
2284 ):
2285 error = event.callback_timed_out_details.error.payload
2287 operation = replace(
2288 operation,
2289 callback_details=CallbackDetails(
2290 callback_id=callback_id,
2291 result=result,
2292 error=error,
2293 ),
2294 )
2296 # STEP details - only update if this event type has result data
2297 if operation_type == OperationType.STEP and event_config.has_result:
2298 existing_step: StepDetails | None = operation.step_details
2299 result_val: str | None = existing_step.result if existing_step else None
2300 error_val: ErrorObject | None = (
2301 existing_step.error if existing_step else None
2302 )
2303 attempt: int = existing_step.attempt if existing_step else 0
2304 next_attempt_ts: datetime.datetime | None = (
2305 existing_step.next_attempt_timestamp if existing_step else None
2306 )
2308 # StepSucceeded provides result
2309 if event.step_succeeded_details:
2310 if event.step_succeeded_details.result:
2311 result_val = event.step_succeeded_details.result.payload
2312 if event.step_succeeded_details.retry_details:
2313 attempt = event.step_succeeded_details.retry_details.current_attempt
2315 # StepFailed provides error and retry details
2316 if event.step_failed_details:
2317 if event.step_failed_details.error:
2318 error_val = event.step_failed_details.error.payload
2319 if event.step_failed_details.retry_details:
2320 attempt = event.step_failed_details.retry_details.current_attempt
2321 if (
2322 event.step_failed_details.retry_details.next_attempt_delay_seconds
2323 is not None
2324 ):
2325 next_attempt_ts = event.event_timestamp + datetime.timedelta(
2326 seconds=event.step_failed_details.retry_details.next_attempt_delay_seconds
2327 )
2329 operation = replace(
2330 operation,
2331 step_details=StepDetails(
2332 result=result_val,
2333 error=error_val,
2334 attempt=attempt,
2335 next_attempt_timestamp=next_attempt_ts,
2336 ),
2337 )
2339 # WAIT details
2340 if operation_type == OperationType.WAIT and event.wait_started_details:
2341 operation = replace(
2342 operation,
2343 wait_details=WaitDetails(
2344 scheduled_end_timestamp=event.wait_started_details.scheduled_end_timestamp
2345 ),
2346 )
2348 # CONTEXT details - only update if this event type has result data (matching TypeScript hasResult)
2349 if operation_type == OperationType.CONTEXT and event_config.has_result:
2350 if (
2351 event.context_succeeded_details
2352 and event.context_succeeded_details.result
2353 ):
2354 operation = replace(
2355 operation,
2356 context_details=ContextDetails(
2357 result=event.context_succeeded_details.result.payload,
2358 error=None,
2359 ),
2360 )
2361 elif event.context_failed_details and event.context_failed_details.error:
2362 operation = replace(
2363 operation,
2364 context_details=ContextDetails(
2365 result=None,
2366 error=event.context_failed_details.error.payload,
2367 ),
2368 )
2370 # CHAINED_INVOKE details - only update if this event type has result data (matching TypeScript hasResult)
2371 if operation_type == OperationType.CHAINED_INVOKE and event_config.has_result:
2372 if (
2373 event.chained_invoke_succeeded_details
2374 and event.chained_invoke_succeeded_details.result
2375 ):
2376 operation = replace(
2377 operation,
2378 chained_invoke_details=ChainedInvokeDetails(
2379 result=event.chained_invoke_succeeded_details.result.payload,
2380 error=None,
2381 ),
2382 )
2383 elif (
2384 event.chained_invoke_failed_details
2385 and event.chained_invoke_failed_details.error
2386 ):
2387 operation = replace(
2388 operation,
2389 chained_invoke_details=ChainedInvokeDetails(
2390 result=None,
2391 error=event.chained_invoke_failed_details.error.payload,
2392 ),
2393 )
2395 # Store in map
2396 operations_map[event.operation_id] = operation
2398 return list(operations_map.values())
2401@dataclass(frozen=True)
2402class GetDurableExecutionHistoryResponse:
2403 """Response containing durable execution history events."""
2405 events: list[Event]
2406 next_marker: str | None = None
2408 @classmethod
2409 def from_dict(cls, data: Mapping[str, Any]) -> GetDurableExecutionHistoryResponse:
2410 events = [Event.from_dict(event_data) for event_data in data.get("Events", [])]
2411 return cls(
2412 events=events,
2413 next_marker=data.get("NextMarker"),
2414 )
2416 def to_dict(self) -> dict[str, Any]:
2417 result: dict[str, Any] = {"Events": [event.to_dict() for event in self.events]}
2418 if self.next_marker is not None:
2419 result["NextMarker"] = self.next_marker
2420 return result
2423class _ExecutionResultSource(Protocol):
2424 operations: list[Operation]
2425 result: DurableExecutionInvocationOutput | None
2428@dataclass(frozen=True)
2429class DurableFunctionTestResult:
2430 status: InvocationStatus
2431 operations: list[Operation]
2432 result: OperationPayload | None = None
2433 error: ErrorObject | None = None
2434 _all_operations: list[Operation] = field(
2435 default_factory=list,
2436 repr=False,
2437 compare=False,
2438 )
2440 @classmethod
2441 def create(cls, execution: _ExecutionResultSource) -> DurableFunctionTestResult:
2442 operations = []
2443 for operation in execution.operations:
2444 if operation.operation_type is OperationType.EXECUTION:
2445 # don't want the EXECUTION operations in the list test code asserts against
2446 continue
2448 if operation.parent_id is None:
2449 operations.append(operation)
2451 if execution.result is None:
2452 msg: str = "Execution result must exist to create test result."
2453 raise DurableFunctionsTestError(msg)
2455 return cls(
2456 status=execution.result.status,
2457 operations=operations,
2458 result=execution.result.result,
2459 error=execution.result.error,
2460 _all_operations=execution.operations,
2461 )
2463 @classmethod
2464 def from_execution_history(
2465 cls,
2466 execution_response: GetDurableExecutionResponse,
2467 history_response: GetDurableExecutionHistoryResponse,
2468 ) -> DurableFunctionTestResult:
2469 """Create test result from execution history responses.
2471 Factory method for cloud runner that builds DurableFunctionTestResult
2472 from GetDurableExecution and GetDurableExecutionHistory API responses.
2473 """
2474 # Map status string to InvocationStatus enum
2475 try:
2476 status = InvocationStatus[execution_response.status]
2477 except KeyError:
2478 logger.warning(
2479 "Unknown status: %s, defaulting to FAILED", execution_response.status
2480 )
2481 status = InvocationStatus.FAILED
2483 # Convert Events to Operations - group by operation_id and merge
2484 try:
2485 svc_operations = events_to_operations(history_response.events)
2486 except Exception as e:
2487 logger.warning("Failed to convert events to operations: %s", e)
2488 svc_operations = []
2490 # Build top-level operation list (exclude EXECUTION type)
2491 operations = []
2492 for svc_op in svc_operations:
2493 if svc_op.operation_type == OperationType.EXECUTION:
2494 continue
2495 if svc_op.parent_id is None:
2496 operations.append(svc_op)
2498 return cls(
2499 status=status,
2500 operations=operations,
2501 result=execution_response.result,
2502 error=execution_response.error,
2503 _all_operations=svc_operations,
2504 )
2506 def get_operation_by_name(self, name: str) -> Operation:
2507 for operation in self.operations:
2508 if operation.name == name:
2509 return operation
2510 msg: str = f"Operation with name '{name}' not found"
2511 raise DurableFunctionsTestError(msg)
2513 def get_step(self, name: str) -> Operation:
2514 return self._get_operation_by_name_and_type(name, OperationType.STEP)
2516 def get_wait(self, name: str) -> Operation:
2517 return self._get_operation_by_name_and_type(name, OperationType.WAIT)
2519 def get_context(self, name: str) -> Operation:
2520 return self._get_operation_by_name_and_type(name, OperationType.CONTEXT)
2522 def get_callback(self, name: str) -> Operation:
2523 return self._get_operation_by_name_and_type(name, OperationType.CALLBACK)
2525 def get_invoke(self, name: str) -> Operation:
2526 return self._get_operation_by_name_and_type(name, OperationType.CHAINED_INVOKE)
2528 def get_execution(self, name: str) -> Operation:
2529 return self._get_operation_by_name_and_type(name, OperationType.EXECUTION)
2531 def get_deserialized_result(self, serdes: ExtendedTypeSerDes | None = None) -> Any:
2532 """Return the deserialized execution result."""
2533 return _deserialize_operation_payload(self.result, serdes)
2535 def get_operation_deserialized_result(
2536 self,
2537 operation: Operation,
2538 serdes: ExtendedTypeSerDes | None = None,
2539 ) -> Any:
2540 """Return the deserialized result payload for a service operation."""
2541 match operation.operation_type:
2542 case OperationType.CONTEXT:
2543 result = (
2544 operation.context_details.result
2545 if operation.context_details
2546 else None
2547 )
2548 case OperationType.STEP:
2549 result = (
2550 operation.step_details.result if operation.step_details else None
2551 )
2552 case OperationType.CALLBACK:
2553 result = (
2554 operation.callback_details.result
2555 if operation.callback_details
2556 else None
2557 )
2558 case OperationType.CHAINED_INVOKE:
2559 result = (
2560 operation.chained_invoke_details.result
2561 if operation.chained_invoke_details
2562 else None
2563 )
2564 case _:
2565 result = None
2566 return _deserialize_operation_payload(result, serdes)
2568 def get_child_operations(self, operation: Operation) -> list[Operation]:
2569 """Return direct child operations for a service operation."""
2570 return [
2571 candidate
2572 for candidate in self._operation_source()
2573 if candidate.parent_id == operation.operation_id
2574 ]
2576 def get_all_operations(self) -> list[Operation]:
2577 """Return all non-execution operations, including nested operations."""
2578 return [
2579 operation
2580 for operation in self._operation_source()
2581 if operation.operation_type != OperationType.EXECUTION
2582 ]
2584 def _operation_source(self) -> list[Operation]:
2585 return self._all_operations or self.operations
2587 def _get_operation_by_name_and_type(
2588 self, name: str, operation_type: OperationType
2589 ) -> Operation:
2590 operation = self.get_operation_by_name(name)
2591 if operation.operation_type != operation_type:
2592 msg = (
2593 f"Operation with name '{name}' has type "
2594 f"{operation.operation_type}, expected {operation_type}"
2595 )
2596 raise DurableFunctionsTestError(msg)
2597 return operation
2600def _deserialize_operation_payload(
2601 payload: OperationPayload | None,
2602 serdes: ExtendedTypeSerDes | None = None,
2603) -> Any:
2604 """Deserialize an operation payload using the provided or default serializer."""
2605 if not payload:
2606 return None
2608 if serdes is None:
2609 serdes = ExtendedTypeSerDes()
2611 try:
2612 return serdes.deserialize_sync(payload)
2613 except Exception:
2614 return json.loads(payload)
2617def _get_callback_id_from_events(
2618 events: list[Event], name: str | None = None
2619) -> str | None:
2620 """
2621 Get callback ID from execution history for callbacks that haven't completed.
2623 Args:
2624 execution_arn: The ARN of the execution to query.
2625 name: Optional callback name to search for. If not provided, returns the latest callback.
2627 Returns:
2628 The callback ID string for a non-completed callback whose creating
2629 invocation has completed, or None if not found.
2631 Raises:
2632 DurableFunctionsTestError: If the named callback has already succeeded/failed/timed out.
2633 """
2634 callback_started_events = [
2635 event for event in events if event.event_type == "CallbackStarted"
2636 ]
2638 if not callback_started_events:
2639 return None
2641 completed_callback_ids = {
2642 event.event_id
2643 for event in events
2644 if event.event_type
2645 in ["CallbackSucceeded", "CallbackFailed", "CallbackTimedOut"]
2646 }
2648 def is_callback_ready(callback_started_event: Event) -> bool:
2649 for event in events[events.index(callback_started_event) + 1 :]:
2650 if event.event_type == "InvocationCompleted":
2651 return True
2652 return False
2654 if name is not None:
2655 for event in callback_started_events:
2656 if event.name == name:
2657 callback_id = event.event_id
2658 if callback_id in completed_callback_ids:
2659 raise DurableFunctionsTestError(
2660 f"Callback {name} has already completed (succeeded/failed/timed out)"
2661 )
2662 if not is_callback_ready(event):
2663 return None
2664 return (
2665 event.callback_started_details.callback_id
2666 if event.callback_started_details
2667 else None
2668 )
2669 return None
2671 # If name is not provided, find the latest non-completed callback event
2672 active_callbacks = [
2673 event
2674 for event in callback_started_events
2675 if event.event_id not in completed_callback_ids and is_callback_ready(event)
2676 ]
2678 if not active_callbacks:
2679 return None
2681 latest_event = active_callbacks[-1]
2682 return (
2683 latest_event.callback_started_details.callback_id
2684 if latest_event.callback_started_details
2685 else None
2686 )
2689@dataclass(frozen=True)
2690class InvokeResponse:
2691 """Response from invoking a durable function."""
2693 invocation_output: DurableExecutionInvocationOutput
2694 request_id: str