使用AWS Python CDK为Step Function添加条件的相关问题

问题1：实现“Do Nothing”并直接跳转至sfn_step_3 #

Step Functions 里不需要单独的“Do Nothing”步骤，你可以直接把False分支的目标设为sfn_step_3；如果需要保持当前 payload 不变再跳转，也可以用Pass步骤（无操作的占位步骤）。

方案1：直接指向目标步骤 #

choice_step = sfn.Choice(self, "Job Complete?")
choice_step.when(sfn.Condition.string_equals("$.is_query", "True"), sfn_special_step)
choice_step.when(sfn.Condition.string_equals("$.is_query", "False"), sfn_step_3)
# 补充默认分支，避免is_query值不符合时流程中断
choice_step.otherwise(sfn_step_3)

# 串联流程
sfn_step_1.next(sfn_step_2).next(choice_step)
sfn_special_step.next(sfn_step_3)  # 特殊步骤执行完后也要走到sfn_step_3

方案2：用Pass步骤做无操作跳转（适合保留payload场景） #

do_nothing = sfn.Pass(self, "Do Nothing")

choice_step = sfn.Choice(self, "Job Complete?")
choice_step.when(sfn.Condition.string_equals("$.is_query", "True"), sfn_special_step)
choice_step.when(sfn.Condition.string_equals("$.is_query", "False"), do_nothing)
choice_step.otherwise(do_nothing)

# 串联流程
sfn_step_1.next(sfn_step_2).next(choice_step)
do_nothing.next(sfn_step_3)
sfn_special_step.next(sfn_step_3)

问题2：在sfn_step_2的payload中添加"is_query": "True" #

直接在TaskInput.from_object的字典里新增键值对即可，因为你要的是固定字符串"True"，直接写死就行：

修改后的代码：

sfn_step_2 = tasks.LambdaInvoke(self, "check-status",
                                lambda_function=check_status_lambda,
                                output_path="$.Payload",
                                integration_pattern=stepfunctions.IntegrationPattern.WAIT_FOR_TASK_TOKEN,
                                payload=stepfunctions.TaskInput.from_object({
                                    "payload.$": "$",
                                    "execution_id.$": "$$.Execution.Id",
                                    "taskToken": stepfunctions.JsonPath.task_token,
                                    "is_query": "True"  # 新增此行
                                })
                                )

如果后续需要动态传入is_query的值（比如从之前步骤的输出获取），可以用JsonPath引用，示例："is_query.$": "$.source_key"。

内容的提问来源于stack exchange，提问作者nokia mobile