def create_state_machine()

in multipagepdfa2i/multipagepdfa2i_stack.py [0:0]

• 52 lines of code
• 1 McCabe index (conditional complexity)

    def create_state_machine(self, services):

        task_pngextract = aws_stepfunctions_tasks.LambdaInvoke(
            self, "PDF. Conver to PNGs",
            lambda_function = services["lambda"]["pngextract"],
            payload_response_only=True,
            result_path = "$.image_keys"
        )

        task_wrapup = aws_stepfunctions_tasks.LambdaInvoke(
            self, "Wrapup and Clean",
            lambda_function = services["lambda"]["wrapup"]
        )

        iterate_sqs_to_textract = aws_stepfunctions_tasks.SqsSendMessage(
            self, "Perform Textract and A2I",
            queue=services["textract_sqs"], 
            message_body = aws_stepfunctions.TaskInput.from_object({
                "token": aws_stepfunctions.Context.task_token,
                "id.$": "$.id",
                "bucket.$": "$.bucket",
                "key.$": "$.key",
                "wip_key.$": "$.wip_key"
            }),
            delay= None,
            integration_pattern=aws_stepfunctions.ServiceIntegrationPattern.WAIT_FOR_TASK_TOKEN
        )

        process_map = aws_stepfunctions.Map(
            self, "Process_Map",
            items_path = "$.image_keys",
            result_path="DISCARD",
            parameters = {
                "id.$": "$.id",
                "bucket.$": "$.bucket",
                "key.$": "$.key",
                "wip_key.$": "$$.Map.Item.Value"
            }
        ).iterator(iterate_sqs_to_textract)
        
        choice_pass = aws_stepfunctions.Pass(
            self,
            "Image. Passing.",
            result=aws_stepfunctions.Result.from_array(["single_image"]),
            result_path="$.image_keys"
        )

        pdf_or_image_choice = aws_stepfunctions.Choice(self, "PDF or Image?")
        pdf_or_image_choice.when(aws_stepfunctions.Condition.string_equals("$.extension", "pdf"), task_pngextract)
        pdf_or_image_choice.when(aws_stepfunctions.Condition.string_equals("$.extension", "png"), choice_pass)
        pdf_or_image_choice.when(aws_stepfunctions.Condition.string_equals("$.extension", "jpg"), choice_pass)

        # Creates the Step Functions
        multipagepdfa2i_sf = aws_stepfunctions.StateMachine(
            scope = self, 
            id = "multipagepdfa2i_stepfunction",
            state_machine_name = "multipagepdfa2i_stepfunction",
            definition=pdf_or_image_choice.afterwards().next(process_map).next(task_wrapup)
        )

        return multipagepdfa2i_sf