How can I use a Task State multiple times in AWS Step Functions using the CDK?

user_19300370 0
Viewed 156

For example I have 3 lambdas :

    sfn_step_1 = tasks.LambdaInvoke(self, "1st-one",
                                            lambda_function=1st-one ,
                                            output_path="$.Payload",
                                            payload=stepfunctions.TaskInput.from_object({
                                                "payload.$": "$",
                                    
                                            })
                                            )
    
     sfn_step_2 = tasks.LambdaInvoke(self, "2nd-One",
                                            lambda_function=2nd-One,
                                            output_path="$.Payload",
                                            payload=stepfunctions.TaskInput.from_object({
                                                "payload.$": "$",
                                     
                                            })
                                            )
    
     sfn_step_3 = tasks.LambdaInvoke(self, "3rd-one",
                                            lambda_function=3rd-one,
                                            output_path="$.Payload",
                                            payload=stepfunctions.TaskInput.from_object({
                                                "payload.$": "$",
                                          
                                            })
                                            )

I want to reuse states 2 and 3 each two times: 1 -> 2 -> 3 -> 2 -> 3

So I am trying to do this:

     state_machine_definition = sfn_step_1.next(sfn_step_2).next(sfn_step_3).next(sfn_step_2).\
                next(sfn_step_3)

but I'm getting this issue:

jsii.errors.JSIIError: State '2nd-One' already has a next state
aws-step-functions aws-cdk aws-lambda amazon-web-services python
1 Answers

You are essentially telling Step Functions to loop infinitely between Steps 2 and 3. That's invalid. Hence the error: Task States can only have one "next" (outbound) path. Fortunately, you can solve your problem simply, without recursion.*

Write a factory function that returns a Lambda Task, passing an id and the function reference. Call the factory function multiple times to build the state machine. This gives you DRY repeating tasks without recursion.

# Task Factory
def make_step(id: str, func: _lambda.Function) -> tasks.LambdaInvoke:
    return tasks.LambdaInvoke(
        self,
        id,
        lambda_function=func,
        payload_response_only=True,
        result_path=f"$.{id}",
    )

sfn.StateMachine(
      self,
      "RepeatStepsMachine",
      definition=make_step("step_1", 1st_One)
      .next(make_step("step_2", 2nd_One))
      .next(make_step("step_3", 3rd_One))
      .next(make_step("step_2b", 2nd_One))
      .next(make_step("step_3b", 3rd_One).next(sfn.Succeed(self, "Done"))),
  )

* If required, recursion can be implemented with a Choice State. Choice States are designed to have multiple "outbound" paths. It's not the best fit for your use case, though.

user_11035110
Related