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
|
import abc
from typing import Final
from moto.stepfunctions.parser.api import (
EvaluationFailedEventDetails,
HistoryEventType,
)
from moto.stepfunctions.parser.asl.component.common.error_name.failure_event import (
FailureEvent,
FailureEventException,
)
from moto.stepfunctions.parser.asl.component.common.error_name.states_error_name import (
StatesErrorName,
)
from moto.stepfunctions.parser.asl.component.common.error_name.states_error_name_type import (
StatesErrorNameType,
)
from moto.stepfunctions.parser.asl.component.common.jsonata.jsonata_template_value_array import (
JSONataTemplateValueArray,
)
from moto.stepfunctions.parser.asl.component.common.string.string_expression import (
StringJSONata,
)
from moto.stepfunctions.parser.asl.component.eval_component import EvalComponent
from moto.stepfunctions.parser.asl.eval.environment import Environment
from moto.stepfunctions.parser.asl.eval.event.event_detail import EventDetails
from moto.stepfunctions.parser.asl.utils.encoding import to_json_str
class Items(EvalComponent, abc.ABC): ...
class ItemsArray(Items):
jsonata_template_value_array: Final[JSONataTemplateValueArray]
def __init__(self, jsonata_template_value_array: JSONataTemplateValueArray):
super().__init__()
self.jsonata_template_value_array = jsonata_template_value_array
def _eval_body(self, env: Environment) -> None:
self.jsonata_template_value_array.eval(env=env)
class ItemsJSONata(Items):
string_jsonata: Final[StringJSONata]
def __init__(self, string_jsonata: StringJSONata):
self.string_jsonata = string_jsonata
def _eval_body(self, env: Environment) -> None:
self.string_jsonata.eval(env=env)
items = env.stack[-1]
if not isinstance(items, list):
# FIXME: If we pass in a 'function' type, the JSONata lib will return a dict and the
# 'unsupported result type state' wont be reached.
def _get_jsonata_value_type_pair(items) -> tuple[str, str]:
if items is None:
return "null", "null"
elif isinstance(items, (int, float)):
if isinstance(items, bool):
return "true" if items else "false", "boolean"
return items, "number"
elif isinstance(items, str):
return f'"{items}"', "string"
elif isinstance(items, dict):
return to_json_str(items, separators=(",", ":")), "object"
expr = self.string_jsonata.literal_value
if jsonata_pair := _get_jsonata_value_type_pair(items):
jsonata_value, jsonata_type = jsonata_pair
error_cause = (
f"The JSONata expression '{expr}' specified for the field 'Items' returned an unexpected result type. "
f"Expected 'array', but was '{jsonata_type}' for value: {jsonata_value}"
)
else:
error_cause = f"The JSONata expression '{expr}' for the field 'Items' returned an unsupported result type."
error_name = StatesErrorName(
typ=StatesErrorNameType.StatesQueryEvaluationError
)
failure_event = FailureEvent(
env=env,
error_name=error_name,
event_type=HistoryEventType.EvaluationFailed,
event_details=EventDetails(
evaluationFailedEventDetails=EvaluationFailedEventDetails(
error=error_name.error_name, cause=error_cause, location="Items"
)
),
)
raise FailureEventException(failure_event=failure_event)
|
