def _create_policy_statements()

in aws_emr_launch/constructs/step_functions/emr_tasks.py [0:0]

32 lines of code
2 McCabe index (conditional complexity)


    def _create_policy_statements(self) -> List[iam.PolicyStatement]:
        stack = core.Stack.of(self)

        policy_statements = list()

        policy_statements.append(
            iam.PolicyStatement(
                effect=iam.Effect.ALLOW,
                actions=["states:StartExecution"],
                resources=[self._state_machine.state_machine_arn],
            )
        )

        if self._integration_pattern == sfn.IntegrationPattern.RUN_JOB:
            policy_statements.append(
                iam.PolicyStatement(
                    effect=iam.Effect.ALLOW,
                    actions=["states:DescribeExecution", "states:StopExecution"],
                    resources=["*"],
                )
            )

            policy_statements.append(
                iam.PolicyStatement(
                    effect=iam.Effect.ALLOW,
                    actions=["events:PutTargets", "events:PutRule", "events:DescribeRule"],
                    resources=[
                        stack.format_arn(
                            service="events",
                            resource="rule",
                            resource_name="StepFunctionsGetEventsForStepFunctionsExecutionRule",
                        )
                    ],
                )
            )

        return policy_statements