Skip to content

StepFunctions, improve the use of ItemsPath declarations in Distributed Map states #12117

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Jan 9, 2025
Merged

StepFunctions, improve the use of ItemsPath declarations in Distributed Map states #12117

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
985870d
evaluate items path iff the input is the previous state output
MEPalma Jan 9, 2025
c71d919
Merge branch 'master' into MEP-sfn-fix_items_path_in_distributed
MEPalma Jan 9, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 10 additions & 2 deletions ...alstack/services/stepfunctions/asl/component/state/state_execution/state_map/state_map.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,9 @@
from localstack.services.stepfunctions.asl.component.state.state_execution.state_map.items.items import (
Items,
)
from localstack.services.stepfunctions.asl.component.state.state_execution.state_map.iteration.distributed_iteration_component import (
DistributedIterationComponent,
)
from localstack.services.stepfunctions.asl.component.state.state_execution.state_map.iteration.itemprocessor.distributed_item_processor import (
DistributedItemProcessor,
DistributedItemProcessorEvalInput,
Expand Down Expand Up @@ -176,8 +179,13 @@ def _eval_execution(self, env: Environment) -> None:
frame.stack = copy.deepcopy(env.stack)

try:
if self.items_path:
self.items_path.eval(env=env)
# ItemsPath in DistributedMap states is only used if a JSONinput is passed from the previous state.
if (
not isinstance(self.iteration_component, DistributedIterationComponent)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

sidenote: complexity is with instance check is probably something to keep an eye on in the future if this would become a common pattern, but it's good to learn about such special cases first :)

or self.item_reader is None
):
if self.items_path:
self.items_path.eval(env=env)

if self.items:
self.items.eval(env=env)
Expand Down
6 changes: 6 additions & 0 deletions tests/aws/services/stepfunctions/templates/scenarios/scenarios_templates.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,9 @@ class ScenariosTemplate(TemplateLoader):
MAP_STATE_CONFIG_DISTRIBUTED_ITEM_SELECTOR: Final[str] = os.path.join(
_THIS_FOLDER, "statemachines/map_state_config_distributed_item_selector.json5"
)
MAP_STATE_CONFIG_DISTRIBUTED_ITEMS_PATH_FROM_PREVIOUS: Final[str] = os.path.join(
_THIS_FOLDER, "statemachines/map_state_config_distributed_items_path_from_previous.json5"
)
MAP_STATE_CONFIG_DISTRIBUTED_ITEM_SELECTOR_PARAMETERS: Final[str] = os.path.join(
_THIS_FOLDER, "statemachines/map_state_config_distributed_item_selector_parameters.json5"
)
Expand Down Expand Up @@ -112,6 +115,9 @@ class ScenariosTemplate(TemplateLoader):
MAP_ITEM_READER_BASE_JSON_MAX_ITEMS_JSONATA: Final[str] = os.path.join(
_THIS_FOLDER, "statemachines/map_item_reader_base_json_max_items_jsonata.json5"
)
MAP_ITEM_READER_BASE_JSON_WITH_ITEMS_PATH: Final[str] = os.path.join(
_THIS_FOLDER, "statemachines/map_item_reader_base_json_with_items_path.json5"
)
MAP_ITEM_BATCHER_BASE_JSON_MAX_PER_BATCH_JSONATA: Final[str] = os.path.join(
_THIS_FOLDER, "statemachines/map_item_batcher_base_max_per_batch_jsonata.json5"
)
Expand Down
43 changes: 43 additions & 0 deletions ...nctions/templates/scenarios/statemachines/map_item_reader_base_json_with_items_path.json5
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,43 @@
{
"StartAt": "LoadState",
"States": {
"LoadState": {
"Type": "Pass",
"Parameters": {
"Bucket.$": "$.Bucket",
"Key.$": "$.Key",
"from_previous": ["from-previous-item-0"]
},
"Next": "MapState"
},
"MapState": {
"Type": "Map",
"MaxConcurrency": 1,
"ItemsPath": "$.from_previous",
"ItemReader": {
"Resource": "arn:aws:states:::s3:getObject",
"Parameters": {
"Bucket.$": "$.Bucket",
"Key.$": "$.Key"
},
"ReaderConfig": {
"InputType": "JSON"
}
},
"ItemProcessor": {
"ProcessorConfig": {
"Mode": "DISTRIBUTED",
"ExecutionType": "STANDARD"
},
"StartAt": "PassState",
"States": {
"PassState": {
"Type": "Pass",
"End": true
}
}
},
"End": true
}
}
}
29 changes: 29 additions & 0 deletions ...lates/scenarios/statemachines/map_state_config_distributed_items_path_from_previous.json5
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,29 @@
{
"StartAt": "PreviousState",
"States": {
"PreviousState": {
"Type": "Pass",
"Result": { "result_value": ["item-value-from-previous"] },
"Next": "MapState"
},
"MapState": {
"Type": "Map",
"MaxConcurrency": 1,
"ItemsPath": "$.result_value",
"ItemProcessor": {
"ProcessorConfig": {
"Mode": "DISTRIBUTED",
"ExecutionType": "STANDARD"
},
"StartAt": "PassState",
"States": {
"PassState": {
"Type": "Pass",
"End": true
}
}
},
"End": true
}
}
}
68 changes: 68 additions & 0 deletions tests/aws/services/stepfunctions/v2/scenarios/test_base_scenarios.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2056,6 +2056,74 @@ def test_map_item_reader_base_json(
exec_input,
)

@markers.aws.validated
@pytest.mark.parametrize(
"items_path",
[
"$.from_previous",
"$[0]",
"$.no_such_path_in_bucket_result",
],
ids=[
"VALID_ITEMS_PATH_FROM_PREVIOUS",
"VALID_ITEMS_PATH_FROM_ITEM_READER",
"INVALID_ITEMS_PATH",
],
)
@markers.snapshot.skip_snapshot_verify(paths=["$..previousEventId"])
def test_map_item_reader_base_json_with_items_path(
self,
aws_client,
s3_create_bucket,
create_state_machine_iam_role,
create_state_machine,
sfn_snapshot,
items_path,
):
bucket_name = s3_create_bucket()
sfn_snapshot.add_transformer(RegexTransformer(bucket_name, "bucket-name"))

key = "file.json"
json_file = json.dumps([["from-bucket-item-0"]])
aws_client.s3.put_object(Bucket=bucket_name, Key=key, Body=json_file)

template = ST.load_sfn_template(ST.MAP_ITEM_READER_BASE_JSON_WITH_ITEMS_PATH)
template["States"]["MapState"]["ItemsPath"] = items_path
definition = json.dumps(template)

exec_input = json.dumps(
{"Bucket": bucket_name, "Key": key, "from_input_items": ["input-item-0"]}
)
create_and_record_execution(
aws_client,
create_state_machine_iam_role,
create_state_machine,
sfn_snapshot,
definition,
exec_input,
)

@markers.aws.validated
@markers.snapshot.skip_snapshot_verify(paths=["$..previousEventId"])
def test_map_state_config_distributed_items_path_from_previous(
self,
aws_client,
create_state_machine_iam_role,
create_state_machine,
sfn_snapshot,
):
template = ST.load_sfn_template(ST.MAP_STATE_CONFIG_DISTRIBUTED_ITEMS_PATH_FROM_PREVIOUS)
definition = json.dumps(template)
exec_input = json.dumps({})
create_and_record_execution(
aws_client,
create_state_machine_iam_role,
create_state_machine,
sfn_snapshot,
definition,
exec_input,
)

@markers.aws.validated
def test_map_item_reader_json_no_json_list_object(
self,
Expand Down
Loading
Loading