def _create_data_firehose_role()

in Gems/AWSMetrics/cdk/aws_metrics/batch_processing.py [0:0]

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

    def _create_data_firehose_role(self) -> None:
        """
        Generated IAM role for the Kinesis Data Firehose delivery stream.
        """
        policy_statements = list()

        data_lake_policy_statement = iam.PolicyStatement(
            actions=[
                's3:AbortMultipartUpload',
                's3:GetBucketLocation',
                's3:GetObject',
                's3:ListBucket',
                's3:ListBucketMultipartUploads',
                's3:PutObject'
            ],
            effect=iam.Effect.ALLOW,
            resources=[
                self._analytics_bucket_arn,
                f'{self._analytics_bucket_arn}/*'
            ]
        )
        policy_statements.append(data_lake_policy_statement)

        events_processing_lambda_policy_statement = iam.PolicyStatement(
            actions=[
                'lambda:InvokeFunction',
                'lambda:GetFunctionConfiguration',
            ],
            effect=iam.Effect.ALLOW,
            resources=[
                self._events_processing_lambda.function_arn
            ]
        )
        policy_statements.append(events_processing_lambda_policy_statement)

        input_stream_policy_statement = iam.PolicyStatement(
            actions=[
                'kinesis:DescribeStream',
                'kinesis:GetShardIterator',
                'kinesis:GetRecords',
                'kinesis:ListShards'
            ],
            effect=iam.Effect.ALLOW,
            resources=[
                self._input_stream_arn
            ]
        )
        policy_statements.append(input_stream_policy_statement)

        log_policy_statement = iam.PolicyStatement(
            actions=[
                'logs:PutLogEvents',
            ],
            effect=iam.Effect.ALLOW,
            resources=[
                self._firehose_delivery_stream_log_group.log_group_arn
            ]
        )
        policy_statements.append(log_policy_statement)

        data_catalog_policy_statement = iam.PolicyStatement(
            actions=[
                'glue:GetTable',
                'glue:GetTableVersion',
                'glue:GetTableVersions'
            ],
            effect=iam.Effect.ALLOW,
            resources=[
                Fn.sub(
                    'arn:${AWS::Partition}:glue:${AWS::Region}:${AWS::AccountId}:catalog'
                ),
                Fn.sub(
                    body='arn:${AWS::Partition}:glue:${AWS::Region}:${AWS::AccountId}:table/${EventsDatabase}/*',
                    variables={
                        'EventsDatabase': self._events_database_name
                    }
                ),
                Fn.sub(
                    body='arn:${AWS::Partition}:glue:${AWS::Region}:${AWS::AccountId}:database/${EventsDatabase}',
                    variables={
                        'EventsDatabase': self._events_database_name
                    }
                )
            ]
        )
        policy_statements.append(data_catalog_policy_statement)

        firehose_delivery_policy = iam.PolicyDocument(
            statements=policy_statements
        )

        self._firehose_delivery_stream_role = iam.Role(
            self._stack,
            id='GameEventsFirehoseRole',
            role_name=resource_name_sanitizer.sanitize_resource_name(
                f'{self._stack.stack_name}-GameEventsFirehoseRole', 'iam_role'),
            assumed_by=iam.ServicePrincipal(
                service='firehose.amazonaws.com'
            ),
            inline_policies={
                'FirehoseDelivery': firehose_delivery_policy
            }
        )