1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
|
from typing import Final, Optional
from moto.stepfunctions.parser.api import (
EvaluationFailedEventDetails,
ExecutionFailedEventDetails,
HistoryEventType,
)
from moto.stepfunctions.parser.asl.component.common.error_name.error_name import (
ErrorName,
)
from moto.stepfunctions.parser.asl.component.common.error_name.states_error_name_type import (
StatesErrorNameType,
)
from moto.stepfunctions.parser.asl.eval.environment import Environment
from moto.stepfunctions.parser.asl.eval.event.event_detail import EventDetails
class FailureEvent:
state_name: Final[str]
source_event_id: Final[int]
error_name: Final[Optional[ErrorName]]
event_type: Final[HistoryEventType]
event_details: Final[Optional[EventDetails]]
def __init__(
self,
env: Environment,
error_name: Optional[ErrorName],
event_type: HistoryEventType,
event_details: Optional[EventDetails] = None,
):
self.state_name = env.next_state_name
self.source_event_id = env.event_history_context.source_event_id
self.error_name = error_name
self.event_type = event_type
self.event_details = event_details
class FailureEventException(Exception):
failure_event: Final[FailureEvent]
def __init__(self, failure_event: FailureEvent):
self.failure_event = failure_event
def extract_error_cause_pair(self) -> Optional[tuple[Optional[str], Optional[str]]]:
if self.failure_event.event_details is None:
return None
failure_event_spec = list(self.failure_event.event_details.values())[0]
error = None
cause = None
if "error" in failure_event_spec:
error = failure_event_spec["error"]
if "cause" in failure_event_spec:
cause = failure_event_spec["cause"]
return error, cause
def get_evaluation_failed_event_details(
self,
) -> Optional[EvaluationFailedEventDetails]:
original_failed_event_details = self.failure_event.event_details[
"evaluationFailedEventDetails"
]
evaluation_failed_event_details = EvaluationFailedEventDetails()
error = original_failed_event_details["error"]
cause = original_failed_event_details["cause"]
location = original_failed_event_details.get("location")
state_name = self.failure_event.state_name
if error != StatesErrorNameType.StatesQueryEvaluationError.to_name():
return None
if error:
evaluation_failed_event_details["error"] = error
if cause:
event_id = self.failure_event.source_event_id
decorated_cause = f"An error occurred while executing the state '{state_name}' (entered at the event id #{event_id}). {cause}"
evaluation_failed_event_details["cause"] = decorated_cause
if location:
evaluation_failed_event_details["location"] = location
if state_name:
evaluation_failed_event_details["state"] = state_name
return evaluation_failed_event_details
def get_execution_failed_event_details(
self,
) -> Optional[ExecutionFailedEventDetails]:
maybe_error_cause_pair = self.extract_error_cause_pair()
if maybe_error_cause_pair is None:
return None
execution_failed_event_details = ExecutionFailedEventDetails()
error, cause = maybe_error_cause_pair
if error:
execution_failed_event_details["error"] = error
if cause:
if (
error == StatesErrorNameType.StatesRuntime.to_name()
or error == StatesErrorNameType.StatesQueryEvaluationError.to_name()
):
state_name = self.failure_event.state_name
event_id = self.failure_event.source_event_id
decorated_cause = f"An error occurred while executing the state '{state_name}' (entered at the event id #{event_id}). {cause}"
execution_failed_event_details["cause"] = decorated_cause
else:
execution_failed_event_details["cause"] = cause
return execution_failed_event_details
|
