DEV Community

Cover image for Create AWS Lambda Function Triggered By S3 Notification Event
πŸš€ Vu Dao πŸš€
πŸš€ Vu Dao πŸš€

Posted on

Create AWS Lambda Function Triggered By S3 Notification Event

  • The Amazon S3 notification feature is able to receive notifications when certain events happen in the bucket.
  • To enable notifications, it must first adds a notification configuration that identifies the events which want Amazon S3 to publish and the destinations where to send the notifications here is lambda function
  • Overview of notifications
  • This post describe how to use AWS chalice to create this.

Alt Text

1. Create aws chalice new-project cdn-invalidation

⚑ $ chalice new-project cdn-invalidation
⚑ $ ls cdn-invalidation  requirements.txt
Enter fullscreen mode Exit fullscreen mode

2. Define which region to create lambda function instead of the default in local aws configuration

⚑ $ export AWS_DEFAULT_REGION=us-east-1
Enter fullscreen mode Exit fullscreen mode

3. Create lamdba function handler

  • The handler listen to s3:ObjectCreated:Put event so any changes in s3://mybucket/static/src will trigger the lambda function Alt Text
from chalice import Chalice
import boto3
import time

app_name = 'cdn-invalidation'
app = Chalice(app_name=app_name)
app.debug = True

class InvalidateCDN:
    """ Invalidate CDN """
    def __init__(self):
        self.distribution_id = 'A1AA1AA11A11AA'
        self.client = boto3.client('cloudfront')

    def create_invalidation(self, file_change):
            res = self.client.create_invalidation(
                    'Paths': {
                        'Quantity': 1,
                        'Items': ["/{}".format(file_change)]
                    'CallerReference': str(time.time()).replace(".", "")
            invalidation_id = res['Invalidation']['Id']
            return invalidation_id
        except Exception as err:
            print(f"Failed to create invalidation, error {err}")

    def get_invalidation_status(self, inval_id):
            res = self.client.get_invalidation(
            return res['Invalidation']['Status']
        except Exception as err:
            print(f"Failed to get invalidation status ID {inval_id}, error {err}")

    def run(self, key):
        print(f"Deploying CDN file: {key}")
        the_id = self.create_invalidation(key)
        count = 0
        while True:
            status = self.get_invalidation_status(the_id)
            if status == 'Completed':
                print(f"Completed, id: {the_id}")
            elif count < 10:
                count += 1
                print("Timeout, please check CDN")

def handle_s3_event(event):
    cdn = InvalidateCDN()
Enter fullscreen mode Exit fullscreen mode

4. Updaterequirements.txt to include boto3 in lambda fuction

⚑ $ cat requirements.txt
Enter fullscreen mode Exit fullscreen mode

5. Deploy

⚑ $ chalice deploy
Enter fullscreen mode Exit fullscreen mode

Alt Text

Alt Text


Top comments (7)

samzoozi profile image
Ali Zonoozi

Is there a way to do this using chalice with CDK integration? when I try deploying using cdk deploy command I get the following error:

Unable to package chalice apps that @app.on_s3_event decorator. CloudFormation does not support modifying the event notifications of existing buckets. You can deploy this app using chalice deploy.

vumdao profile image
πŸš€ Vu Dao πŸš€

Unfortunately I migrated all my chalice projects to fully CDK :)

baotran2207 profile image
baotran2207 • Edited

With current version , we can not use on_s3_event with cdk deploy. The only option i think is s3 event -> sqs/sns -> chalice.on_sqs_message . There is one cons is that we need to filter the prefix/suffix by ourself in on_sqs_message handler .

I assume you follow the cdk deploy tutorial

This is example chalice stack infrastructure/stacks/

from aws_cdk import (
    aws_sns, ## you can notify sns
    aws_sqs, ## you can notify sqs
    aws_s3_notifications as aws_s3_noti,
from chalice.cdk import Chalice
RUNTIME_SOURCE_DIR = os.path.join(
    os.path.dirname(os.path.dirname(__file__)), os.pardir, "runtime"

class ChaliceApp(cdk.Stack):
     def __init__(self, scope, id, **kwargs):
         self.sqs_generic = aws_sqs.Queue(self, "mysqs")
         self.bucket = s3.Bucket(self, "MyBucket")
         self.chalice = Chalice(
                "lambda_memory_size": 256,
                "environment_variables": {
                    "S3_MAIN_BUCKET": self.bucket.bucket_name,
                    "SQS_GENERIC" : self.sqs_generic.queue_name,
         self.chalice_role = self.chalice.get_role("DefaultRole")
          self.bucket.add_event_notification(aws_s3.EventType.OBJECT_CREATED_PUT, aws_s3_noti.SqsDestination(self.sqs_generic))
Enter fullscreen mode Exit fullscreen mode

Then in your runtime/ , you can receive the sqs message :

from chalice import Chalice
app = Chalice(app_name='chalice-backend')
    queue=os.environ.get("SQS_GENERIC", ""),
def handle_sqs_message(event):
    print('Trigger generic')
    print("dict ", event.to_dict())

    for record in event:
        print(record, 'in event')" in even ! Detail {record} ")
    ## with the bucket and the key , you can open boto3 session and implement logic here 

app.register_blueprint(sqs_events) ## remember to register events
Enter fullscreen mode Exit fullscreen mode

And after deployment, upload a file to s3 then in cloudwatch , there would be something like this

Image description

sillians profile image
Basil Ihuoma

I need help... I have this scrapper that gets data from a website, please how do I use lambda function with EventBridge to trigger the scrapper to get data on a daily basis.

vumdao profile image
πŸš€ Vu Dao πŸš€

Use EventBridge with cron schedule to trigger lambda function

sillians profile image
Basil Ihuoma


routinggames profile image
Duy Nguy3n

thanks for sharing