AWS CDK - How to add an event notification to an existing S3 Bucket

Question

I'm trying to modify this AWS-provided CDK example to instead use an existing bucket. Additional documentation indicates that importing existing resources is supported. So far I am unable to add an event notification to the existing bucket using CDK.

Here is my modified version of the example:

class S3TriggerStack(core.Stack):

    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
        super().__init__(scope, id, **kwargs)

        # create lambda function
        function = _lambda.Function(self, "lambda_function",
                                    runtime=_lambda.Runtime.PYTHON_3_7,
                                    handler="lambda-handler.main",
                                    code=_lambda.Code.asset("./lambda"))

        # **MODIFIED TO GET EXISTING BUCKET**
        #s3 = _s3.Bucket(self, "s3bucket")
        s3 = _s3.Bucket.from_bucket_arn(self, 's3_bucket',
            bucket_arn='arn:<my_region>:::<my_bucket>')

        # create s3 notification for lambda function
        notification = aws_s3_notifications.LambdaDestination(function)

        # assign notification for the s3 event type (ex: OBJECT_CREATED)
        s3.add_event_notification(_s3.EventType.OBJECT_CREATED, notification)

This results in the following error when trying to add_event_notification:

AttributeError: '_IBucketProxy' object has no attribute 'add_event_notification'

The from_bucket_arn function returns an IBucket, and the add_event_notification function is a method of the Bucket class, but I can't seem to find any other way to do this. Maybe it's not supported. Any help would be appreciated.

Answer 1

since June 2021 there is a nicer way to solve this problem. Since approx. Version 1.110.0 of the CDK it is possible to use the S3 notifications with Typescript Code:

Example:

const s3Bucket = s3.Bucket.fromBucketName(this, 'bucketId', 'bucketName');
s3Bucket.addEventNotification(s3.EventType.OBJECT_CREATED, new s3n.LambdaDestination(lambdaFunction), {
    prefix: 'example/file.txt'
});

CDK Documentation: https://docs.aws.amazon.com/cdk/api/latest/docs/aws-s3-notifications-readme.html

Pull Request: https://github.com/aws/aws-cdk/pull/15158

Answer 2

UPDATED: Source code from original answer will overwrite existing notification list for bucket which will make it impossible adding new lambda triggers. Here's the solution which uses event sources to handle mentioned problem.

import aws_cdk {
    aws_s3 as s3,
    aws_cdk.aws_lambda as lambda_
    aws_lambda_event_sources as event_src
}
import path as path

class S3LambdaTrigger(core.Stack):
    
    def __init__(self, scope: core.Construct, id: str):
        
        super().__init__(scope, id)
        
        bucket = s3.Bucket(
            self, "S3Bucket",
            block_public_access=s3.BlockPublicAccess.BLOCK_ALL,
            bucket_name='BucketName',
            encryption=s3.BucketEncryption.S3_MANAGED,
            versioned=True
        )

        fn = lambda_.Function(
            self, "LambdaFunction",
            runtime=lambda_.Runtime.NODEJS_10_X,
            handler="index.handler",
            code=lambda_.Code.from_asset(path.join(__dirname, "lambda-handler"))
        )

        fn.add_permission(
            's3-service-principal', 
            principal=aws_iam.ServicePrincipal('s3.amazonaws.com')
        )

        fn.add_event_source(
            event_src.S3EventSource(
                bucket, 
                events=[s3.EventType.OBJECT_CREATED, s3.EventType.OBJECT_REMOVED],
                filters=[s3.NotificationKeyFilter(prefix="subdir/", suffix=".txt")]
            )
        )

ORIGINAL: I took ubi's solution in TypeScript and successfully translated it to Python. His solution worked for me.

#!/usr/bin/env python

from typing import List

from aws_cdk import (
    core,
    custom_resources as cr,
    aws_lambda as lambda_,
    aws_s3 as s3,
    aws_iam as iam,
)


class S3NotificationLambdaProps:
    def __init__(self, bucket: s3.Bucket, function: lambda_.Function, events: List[str], prefix: str):
        self.bucket = bucket
        self.function = function
        self.events = events
        self.prefix = prefix


class S3NotificationLambda(core.Construct):

    def __init__(self, scope: core.Construct, id: str, props: S3NotificationLambdaProps):
        super().__init__(scope, id)

        self.notificationResource = cr.AwsCustomResource(
            self, f'CustomResource{id}',
            on_create=cr.AwsSdkCall(
                service="S3",
                action="S3:putBucketNotificationConfiguration",
                # Always update physical ID so function gets executed
                physical_resource_id=cr.PhysicalResourceId.of(f'S3NotifCustomResource{id}'),
                parameters={
                    "Bucket": props.bucket.bucket_name,
                    "NotificationConfiguration": {
                        "LambdaFunctionConfigurations": [{
                            "Events": props.events,
                            "LambdaFunctionArn": props.function.function_arn,
                            "Filter": {
                                "Key": {"FilterRules": [{"Name": "prefix", "Value": props.prefix}]}
                            }}
                        ]
                    }
                }
            ),
            on_delete=cr.AwsSdkCall(
                service="S3",
                action="S3:putBucketNotificationConfiguration",
                # Always update physical ID so function gets executed
                physical_resource_id=cr.PhysicalResourceId.of(f'S3NotifCustomResource{id}'),
                parameters={
                    "Bucket": props.bucket.bucket_name,
                    "NotificationConfiguration": {},
                }
            ),
            policy=cr.AwsCustomResourcePolicy.from_statements(
                statements=[
                    iam.PolicyStatement(
                        actions=["S3:PutBucketNotification", "S3:GetBucketNotification"],
                        resources=[props.bucket.bucket_arn]
                    ),
                ]
            )
        )

        props.function.add_permission(
            "AllowS3Invocation",
            action="lambda:InvokeFunction",
            principal=iam.ServicePrincipal("s3.amazonaws.com"),
            source_arn=props.bucket.bucket_arn,
        )

        # don't create the notification custom-resource until after both the bucket and lambda
        # are fully created and policies applied.
        self.notificationResource.node.add_dependency(props.bucket)
        self.notificationResource.node.add_dependency(props.function)

# Usage:

s3NotificationLambdaProps = S3NotificationLambdaProps(
    bucket=bucket_,
    function=lambda_fn_,
    events=['s3:ObjectCreated:*'],
    prefix='foo/'
)

s3NotificationLambda = S3NotificationLambda(
    self, "S3NotifLambda",
    self.s3NotificationLambdaProps
)