Event Source Data Classes - Powertools for AWS Lambda (Python) (original) (raw)

Event Source Data Classes provides self-describing and strongly-typed classes for various AWS Lambda event sources.

Key features¶

• Type hinting and code completion for common event types
• Helper functions for decoding/deserializing nested fields
• Docstrings for fields contained in event schemas
• Standardized attribute-based access to event properties

Getting started¶

Tip

All examples shared in this documentation are available within the project repository.

There are two ways to use Event Source Data Classes in your Lambda functions.

Method 1: Direct Initialization

You can initialize the appropriate data class by passing the Lambda event object to its constructor.

app.pyAPI Gateway Proxy Example Event

Method 2: Using the event_source Decorator

Alternatively, you can use the event_source decorator to automatically parse the event.

app.pyAPI Gateway Proxy Example Event

Autocomplete with self-documented properties and methods¶

Event Source Data Classes has the ability to leverage IDE autocompletion and inline documentation. When using the APIGatewayProxyEvent class, for example, the IDE will offer autocomplete suggestions for various properties and methods.

Supported event sources¶

Each event source is linked to its corresponding GitHub file with the full set of properties, methods, and docstrings specific to each event type.

Info

The examples showcase a subset of Event Source Data Classes capabilities - for comprehensive details, leverage your IDE's autocompletion, refer to type hints and docstrings, and explore the full API reference for complete property listings of each event source.

Active MQ¶

It is used for Active MQ payloads, also see the AWS blog postfor more details.

app.pyActive MQ Example Event

It is used for API Gateway Rest API Lambda Authorizer payload.

Use APIGatewayAuthorizerRequestEvent for type REQUEST and APIGatewayAuthorizerTokenEvent for type TOKEN.

Rest APIsWebSocket APIsAPI Gateway Authorizer Request Example Eventapp_token.pyAPI Gateway Authorizer Token Example Event

API Gateway Authorizer V2¶

It is used for API Gateway HTTP API Lambda Authorizer payload version 2. See also this blog post for more details.

app.pyAPI Gateway Authorizer V2 Example Event

API Gateway Proxy¶

It is used for either API Gateway REST API or HTTP API using v1 proxy event.

app.pyAPI Gateway Proxy Example Event

API Gateway Proxy V2¶

It is used for HTTP API using v2 proxy event.

app.pyAPI Gateway Proxy V2 Example Event

Application Load Balancer¶

Is it used for Application load balancer event.

app.pyApplication Load Balancer Example Event

AppSync Authorizer¶

Used when building an AWS_LAMBDA Authorization with AppSync. See blog post Introducing Lambda authorization for AWS AppSync GraphQL APIsor read the Amplify documentation on using AWS Lambda for authorization with AppSync.

app.pyAppSync Authorizer Example Event

AppSync Resolver¶

Used when building Lambda GraphQL Resolvers with Amplify GraphQL Transform Library (@function), and AppSync Direct Lambda Resolvers.

The example serves as an AppSync resolver for the locations field of the Merchant type. It uses the @event_source decorator to parse the AppSync event, handles pagination and filtering for locations, and demonstrates AppSyncIdentityCognito.

app.pyAppSync Resolver Example Event

AWS Config Rule¶

The example utilizes AWSConfigRuleEvent to parse the incoming event. The function logs the message type of the invoking event and returns a simple success response. The example event receives a Scheduled Event Notification, but could also be ItemChanged and Oversized.

app.pyScheduledNotification Example Event

Bedrock Agent¶

The example handles Bedrock Agent event with BedrockAgentEvent to parse the incoming event. The function logs the action group and input text, then returns a structured response compatible with Bedrock Agent's expected format, including a mock response body.

app.pyBedrock Agent Example Event

CloudFormation Custom Resource¶

The example focuses on the Create request type, generating a unique physical resource ID and logging the process. The function is structured to potentially handle Update and Delete operations as well.

app.pyCloudFormation Custom Resource Example Event

CloudWatch Dashboard Custom Widget¶

Thie example for CloudWatchDashboardCustomWidgetEvent logs the dashboard name, extracts key information like widget ID and time range, and returns a formatted response with a title and markdown content. Read more about custom widgets for Cloudwatch dashboard.

app.pyCloudWatch Dashboard Example Event

CloudWatch Alarm State Change Action¶

CloudWatch supports Lambda as an alarm state change action. You can use the CloudWathAlarmEvent data class to access the fields containing such data as alarm information, current state, and previous state.

app.pyCloudWatch Alarm Example Event

CloudWatch Logs¶

CloudWatch Logs events by default are compressed and base64 encoded. You can use the helper function provided to decode, decompress and parse json data from the event.

app.pyCloudWatch Logs Example Event

Kinesis integration¶

When streaming CloudWatch Logs to a Kinesis Data Stream (cross-account or not), you can use extract_cloudwatch_logs_from_event to decode, decompress and extract logs as CloudWatchLogsDecodedData to ease log processing.

app.pyKinesis Stream CloudWatch Logs Example Event

Alternatively, you can use extract_cloudwatch_logs_from_record to seamless integrate with the Batch utility for more robust log processing.

app.pyKinesis Stream CloudWatch Logs Example Event

CodeDeploy LifeCycle Hook¶

CodeDeploy triggers Lambdas with this event when defined inAppSpec definitionsto test applications at different stages of deployment.

app.pyCodeDeploy LifeCycle Hook Example Event

CodePipeline Job¶

Data classes and utility functions to help create continuous delivery pipelines tasks with AWS Lambda.

app.pyCodePipeline Job Example Event

Cognito User Pool¶

Cognito User Pools have several different Lambda trigger sources, all of which map to a different data class, which can be imported from aws_lambda_powertools.data_classes.cognito_user_pool_event:

Some examples for the Cognito User Pools Lambda triggers sources:

Post Confirmation Example¶

app.pyCognito Post Confirmation Example Event

Define Auth Challenge Example¶

Note

In this example we are modifying the wrapped dict response fields, so we need to return the json serializable wrapped event in event.raw_event.

This example is based on the AWS Cognito docs for Define Auth Challenge Lambda Trigger.

app.pyCognito Define Auth Challengen Example Event

Create Auth Challenge Example¶

This example is based on the AWS Cognito docs for Create Auth Challenge Lambda Trigger.

app.pyCognito Create Auth Challengen Example Event

Verify Auth Challenge Response Example¶

This example is based on the AWS Cognito docs for Verify Auth Challenge Response Lambda Trigger.

app.pyCognito Verify Auth Challengen Example Event

Connect Contact Flow¶

The example integrates with Amazon Connect by handling contact flow events. The function converts the event into a ConnectContactFlowEvent object, providing a structured representation of the contact flow data.

app.pyConnect Contact Flow Example Event

DynamoDB Streams¶

The DynamoDB data class utility provides the base class for DynamoDBStreamEvent, as well as enums for stream view type (StreamViewType) and event type. (DynamoDBRecordEventName). The class automatically deserializes DynamoDB types into their equivalent Python types.

app.pyapp_multiple_records.pyDynamoDB Streams Example Event

EventBridge¶

When an event matching a defined rule occurs in EventBridge, it can automatically trigger a Lambda function, passing the event data as input.

app.pyEventBridge Example Event

Kafka¶

This example is based on the AWS docs for Amazon MSK and self-managed Apache Kafka.

app.pyKafka Example Event

Kinesis streams¶

Kinesis events by default contain base64 encoded data. You can use the helper function to access the data either as json or plain text, depending on the original payload.

app.pyKinesis streams Example Event

Kinesis Firehose delivery stream¶

When using Kinesis Firehose, you can use a Lambda function to perform data transformation. For each transformed record, you can choose to either:

• A) Put them back to the delivery stream (default)
• B) Drop them so consumers don't receive them (e.g., data validation)
• C) Indicate a record failed data transformation and should be retried

To do that, you can use KinesisFirehoseDataTransformationResponse class along with helper functions to make it easier to decode and encode base64 data in the stream.

Transforming streaming recordsDropping invalid recordsIndicating a processing failurekinesisFirehoseEvent.json

1. Ingesting JSON payloads?
   Use record.data_as_json to easily deserialize them.
2. For your convenience, base64_from_json serializes a dict to JSON, then encode as base64 data.

1. This exception would be generated from record.data_as_json if invalid payload.

1. This record will now be sent to your S3 bucket in the processing-failed folder.

Lambda Function URL¶

Lambda Function URLs provide a direct HTTP endpoint for invoking Lambda functions. This feature allows functions to receive and process HTTP requests without the need for additional services like API Gateway.

app.pyLambda Function URL Example Event

Rabbit MQ¶

It is used for Rabbit MQ payloads. See the blog postfor more details.

app.pyRabbit MQ Example Event

S3¶

Integration with Amazon S3 enables automatic, serverless processing of object-level events in S3 buckets. When triggered by actions like object creation or deletion, Lambda functions receive detailed event information, allowing for real-time file processing, data transformations, and automated workflows.

app.pyS3 Example Event

S3 Batch Operations¶

This example is based on the AWS S3 Batch Operations documentation Example Lambda function for S3 Batch Operations.

app.pyS3 Batch Operations Example Event

S3 Object Lambda¶

This example is based on the AWS Blog post Introducing Amazon S3 Object Lambda – Use Your Code to Process Data as It Is Being Retrieved from S3.

app.pyS3 Object Lambda Example Event

S3 EventBridge Notification¶

S3 EventBridge notifications enhance Lambda's ability to process S3 events by routing them through Amazon EventBridge. This integration offers advanced filtering, multiple destination support, and standardized CloudEvents format.

app.pyS3 EventBridge Notification Example Event

Secrets Manager¶

AWS Secrets Manager rotation uses an AWS Lambda function to update the secret. Click here for more information about rotating AWS Secrets Manager secrets.

app.pySecrets Manager Example Event

SES¶

The integration with Simple Email Service (SES) enables serverless email processing. When configured, SES can trigger Lambda functions in response to incoming emails or delivery status notifications. The Lambda function receives an SES event containing details like sender, recipients, and email content.

app.pySES Example Event

SNS¶

The integration with Simple Notification Service (SNS) enables serverless message processing. When configured, SNS can trigger Lambda functions in response to published messages or notifications. The Lambda function receives an SNS event containing details like the message body, subject, and metadata.

app.pySNS Example Event

SQS¶

The integration with Simple Queue Service (SQS) enables serverless queue processing. When configured, SQS can trigger Lambda functions in response to messages in the queue. The Lambda function receives an SQS event containing details like message body, attributes, and metadata.

app.pySQS Example Event

VPC Lattice V2¶

You can register your Lambda functions as targets within an Amazon VPC Lattice service network. By doing this, your Lambda function becomes a service within the network, and clients that have access to the VPC Lattice service network can call your service using Payload V2.

Click here for more information about using AWS Lambda with Amazon VPC Lattice.

app.pyLattice Example Event

VPC Lattice V1¶

You can register your Lambda functions as targets within an Amazon VPC Lattice service network. By doing this, your Lambda function becomes a service within the network, and clients that have access to the VPC Lattice service network can call your service.

Click here for more information about using AWS Lambda with Amazon VPC Lattice.

app.pyLattice Example Event

IoT Core Events¶

IoT Core Thing Created/Updated/Deleted¶

You can use IoT Core registry events to trigger your lambda functions. More information on this specific one can be found here.

app.pyExample Event

IoT Core Thing Type Created/Updated/Deprecated/Undeprecated/Deleted¶

You can use IoT Core registry events to trigger your lambda functions. More information on this specific one can be found here.

app.pyExample Event

IoT Core Thing Type Associated/Disassociated with a Thing¶

You can use IoT Core registry events to trigger your lambda functions. More information on this specific one can be found here.

app.pyExample Event

IoT Core Thing Group Created/Updated/Deleted¶

You can use IoT Core registry events to trigger your lambda functions. More information on this specific one can be found here.

app.pyExample Event

IoT Thing Added/Removed from Thing Group¶

You can use IoT Core registry events to trigger your lambda functions. More information on this specific one can be found here.

app.pyExample Event

IoT Child Group Added/Deleted from Parent Group¶

You can use IoT Core registry events to trigger your lambda functions. More information on this specific one can be found here.

app.pyExample Event

Advanced¶

Debugging¶

Alternatively, you can print out the fields to obtain more information. All classes come with a __str__ method that generates a dictionary string which can be quite useful for debugging.

However, certain events may contain sensitive fields such as secret_access_key and session_token, which are labeled as [SENSITIVE] to prevent any accidental disclosure of confidential information.

If we fail to deserialize a field value (e.g., JSON), they will appear as [Cannot be deserialized]

debugging.pydebugging_event.jsondebugging_output.json