def _get_awsbatch_cli_write_policy()

in cli/src/pcluster/templates/awsbatch_builder.py [0:0]

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

    def _get_awsbatch_cli_write_policy(self):
        """Return list of WRITE policies required by ParallelCluster AWS Batch CLI."""
        return iam.PolicyStatement(
            sid="BatchCliWritePermissions",
            actions=[
                "batch:SubmitJob",  # required by awsbsub command
                "batch:TerminateJob",  # required by awsbkill
                "logs:GetLogEvents",  # required by awsbout
                "ecs:ListContainerInstances",  # required by awsbhosts
                "ecs:DescribeContainerInstances",  # required by awsbhosts
                "s3:PutObject",  # required by awsbsub
            ],
            effect=iam.Effect.ALLOW,
            resources=[
                self._format_arn(
                    service="logs",
                    account=self._stack_account,
                    region=self._stack_region,
                    resource="log-group:/aws/batch/job:log-stream:PclusterJobDefinition*",
                ),
                self._format_arn(
                    service="ecs",
                    account=self._stack_account,
                    region=self._stack_region,
                    resource="container-instance/AWSBatch-PclusterComputeEnviron*",
                ),
                self._format_arn(
                    service="ecs",
                    account=self._stack_account,
                    region=self._stack_region,
                    resource="cluster/AWSBatch-Pcluster*",
                ),
                self._format_arn(
                    service="batch",
                    account=self._stack_account,
                    region=self._stack_region,
                    resource="job-queue/PclusterJobQueue*",
                ),
                self._format_arn(
                    service="batch",
                    account=self._stack_account,
                    region=self._stack_region,
                    resource="job-definition/PclusterJobDefinition*:*",
                ),
                self._format_arn(
                    service="batch",
                    account=self._stack_account,
                    region=self._stack_region,
                    resource="job/*",
                ),
                self._format_arn(
                    service="s3",
                    account="",
                    region="",
                    resource=f"{self.bucket.name}/{self.bucket.artifact_directory}/batch/*",
                ),
            ],
        )