tessl/pypi-aws-lambda-powertools
Comprehensive developer toolkit implementing serverless best practices for AWS Lambda functions in Python
89
1.21x
event-handlers.mddocs/
Event Handlers
Framework for building Lambda functions that handle HTTP events from API Gateway, Application Load Balancer, AppSync GraphQL, Bedrock Agents, and other AWS services with automatic request/response serialization, routing, and CORS support.
Capabilities
API Gateway REST Resolver
HTTP request resolver for API Gateway REST APIs with automatic routing, request validation, and response formatting.
class APIGatewayRestResolver:
def __init__(
self,
cors: CORSConfig = None,
debug: bool = None,
serializer: Callable[[Dict], str] = None,
strip_prefixes: List[str] = None,
enable_validation: bool = False,
):
"""
Initialize API Gateway REST resolver.
Parameters:
- cors: CORS configuration for cross-origin requests
- debug: Enable debug mode for detailed error responses
- serializer: Custom JSON serializer function
- strip_prefixes: Path prefixes to strip from routes
- enable_validation: Enable request validation
"""
def get(self, rule: str, **kwargs) -> Callable:
"""
Register GET route handler.
Parameters:
- rule: URL route pattern (supports path parameters)
- **kwargs: Additional route configuration
Returns:
Decorator function for route handler
"""
def post(self, rule: str, **kwargs) -> Callable:
"""Register POST route handler"""
def put(self, rule: str, **kwargs) -> Callable:
"""Register PUT route handler"""
def patch(self, rule: str, **kwargs) -> Callable:
"""Register PATCH route handler"""
def delete(self, rule: str, **kwargs) -> Callable:
"""Register DELETE route handler"""
def head(self, rule: str, **kwargs) -> Callable:
"""Register HEAD route handler"""
def options(self, rule: str, **kwargs) -> Callable:
"""Register OPTIONS route handler"""
def route(
self,
rule: str,
method: Union[str, List[str]],
**kwargs,
) -> Callable:
"""
Register route handler for specific HTTP method(s).
Parameters:
- rule: URL route pattern
- method: HTTP method or list of methods
- **kwargs: Additional route configuration
"""
def resolve(self, event: Dict, context: LambdaContext) -> Dict:
"""
Resolve incoming Lambda event to registered route handler.
Parameters:
- event: API Gateway Lambda proxy integration event
- context: Lambda runtime context
Returns:
API Gateway response dictionary
"""
@property
def current_event(self) -> APIGatewayProxyEvent:
"""Get current request event during handler execution"""
@property
def lambda_context(self) -> LambdaContext:
"""Get Lambda context during handler execution"""
def append_context(self, **kwargs) -> None:
"""Append key-value pairs to request context"""
def clear_context(self) -> None:
"""Clear request context"""
def exception_handler(self, exception_type: Exception) -> Callable:
"""
Register exception handler for specific exception type.
Parameters:
- exception_type: Exception class to handle
Returns:
Decorator function for exception handler
"""
def not_found(self) -> Callable:
"""Register handler for 404 Not Found responses"""
def unauthorized(self) -> Callable:
"""Register handler for 401 Unauthorized responses"""
def include_router(self, router: "APIGatewayRestResolver", prefix: str = None) -> None:
"""
Include routes from another resolver.
Parameters:
- router: Another APIGatewayRestResolver instance
- prefix: Optional path prefix for included routes
"""
def enable_swagger(
self,
path: str = "/swagger",
persist_authorization: bool = True,
**kwargs,
) -> None:
"""
Enable Swagger UI documentation.
Parameters:
- path: Path to serve Swagger UI
- persist_authorization: Whether to persist auth across requests
- **kwargs: Additional Swagger configuration
"""API Gateway HTTP Resolver
HTTP request resolver optimized for API Gateway HTTP APIs (v2.0) with improved performance and simpler payload format.
class APIGatewayHttpResolver:
def __init__(
self,
cors: CORSConfig = None,
debug: bool = None,
serializer: Callable[[Dict], str] = None,
strip_prefixes: List[str] = None,
enable_validation: bool = False,
):
"""
Initialize API Gateway HTTP resolver for v2.0 format.
Parameters:
- cors: CORS configuration
- debug: Enable debug mode
- serializer: Custom JSON serializer
- strip_prefixes: Path prefixes to strip
- enable_validation: Enable request validation
"""
# Inherits same route methods as APIGatewayRestResolver
def get(self, rule: str, **kwargs) -> Callable: ...
def post(self, rule: str, **kwargs) -> Callable: ...
def put(self, rule: str, **kwargs) -> Callable: ...
def delete(self, rule: str, **kwargs) -> Callable: ...
def route(self, rule: str, method: Union[str, List[str]], **kwargs) -> Callable: ...
def resolve(self, event: Dict, context: LambdaContext) -> Dict:
"""Resolve HTTP API v2.0 event to route handler"""
@property
def current_event(self) -> APIGatewayProxyEventV2:
"""Get current HTTP API v2.0 request event"""Application Load Balancer Resolver
HTTP request resolver for Application Load Balancer target group integration.
class ALBResolver:
def __init__(
self,
cors: CORSConfig = None,
debug: bool = None,
serializer: Callable[[Dict], str] = None,
strip_prefixes: List[str] = None,
enable_validation: bool = False,
):
"""
Initialize Application Load Balancer resolver.
Parameters:
- cors: CORS configuration
- debug: Enable debug mode
- serializer: Custom JSON serializer
- strip_prefixes: Path prefixes to strip
- enable_validation: Enable request validation
"""
# Same routing interface as API Gateway resolvers
def get(self, rule: str, **kwargs) -> Callable: ...
def post(self, rule: str, **kwargs) -> Callable: ...
def resolve(self, event: Dict, context: LambdaContext) -> Dict: ...
@property
def current_event(self) -> ALBEvent:
"""Get current ALB request event"""AppSync Resolver
GraphQL resolver for AWS AppSync with support for direct Lambda resolvers and batch resolvers.
class AppSyncResolver:
def __init__(self, debug: bool = None):
"""
Initialize AppSync GraphQL resolver.
Parameters:
- debug: Enable debug mode for detailed error responses
"""
def resolver(
self,
type_name: str = "*",
field_name: str = None,
) -> Callable:
"""
Register GraphQL field resolver.
Parameters:
- type_name: GraphQL type name or "*" for any type
- field_name: GraphQL field name or None for any field
Returns:
Decorator function for field resolver
"""
def batch_resolver(
self,
type_name: str = "*",
field_name: str = None,
) -> Callable:
"""
Register batch GraphQL field resolver for efficient N+1 resolution.
Parameters:
- type_name: GraphQL type name
- field_name: GraphQL field name
Returns:
Decorator function for batch resolver
"""
def resolve(self, event: Dict, context: LambdaContext) -> Any:
"""
Resolve AppSync direct Lambda resolver event.
Parameters:
- event: AppSync resolver event
- context: Lambda runtime context
Returns:
Resolver result (any JSON-serializable value)
"""
@property
def current_event(self) -> AppSyncResolverEvent:
"""Get current AppSync resolver event"""
def append_context(self, **kwargs) -> None:
"""Append context for resolver execution"""AppSync Events Resolver
Event-driven resolver for AppSync subscriptions and real-time updates.
class AppSyncEventsResolver:
def __init__(self, debug: bool = None):
"""
Initialize AppSync Events resolver for subscriptions.
Parameters:
- debug: Enable debug mode
"""
def resolver(
self,
type_name: str = "*",
field_name: str = None,
) -> Callable:
"""Register AppSync Events field resolver"""
def resolve(self, event: Dict, context: LambdaContext) -> Any:
"""Resolve AppSync Events resolver event"""
@property
def current_event(self) -> AppSyncResolverEventsEvent:
"""Get current AppSync Events resolver event"""Lambda Function URL Resolver
HTTP request resolver for Lambda Function URLs with simplified routing.
class LambdaFunctionUrlResolver:
def __init__(
self,
cors: CORSConfig = None,
debug: bool = None,
serializer: Callable[[Dict], str] = None,
strip_prefixes: List[str] = None,
enable_validation: bool = False,
):
"""
Initialize Lambda Function URL resolver.
Parameters:
- cors: CORS configuration
- debug: Enable debug mode
- serializer: Custom JSON serializer
- strip_prefixes: Path prefixes to strip
- enable_validation: Enable request validation
"""
# Same routing interface
def get(self, rule: str, **kwargs) -> Callable: ...
def post(self, rule: str, **kwargs) -> Callable: ...
def resolve(self, event: Dict, context: LambdaContext) -> Dict: ...
@property
def current_event(self) -> LambdaFunctionUrlEvent:
"""Get current Function URL request event"""Bedrock Agent Resolver
Resolver for Amazon Bedrock Agent Lambda functions with action group integration.
class BedrockAgentResolver:
def __init__(self, debug: bool = None):
"""
Initialize Bedrock Agent resolver.
Parameters:
- debug: Enable debug mode
"""
def action(self, name: str) -> Callable:
"""
Register Bedrock Agent action handler.
Parameters:
- name: Action name as defined in agent configuration
Returns:
Decorator function for action handler
"""
def resolve(self, event: Dict, context: LambdaContext) -> BedrockResponse:
"""
Resolve Bedrock Agent event to action handler.
Parameters:
- event: Bedrock Agent invocation event
- context: Lambda runtime context
Returns:
BedrockResponse object
"""
@property
def current_event(self) -> BedrockAgentEvent:
"""Get current Bedrock Agent event"""
class BedrockAgentFunctionResolver:
def __init__(self, debug: bool = None):
"""Initialize Bedrock Agent Function resolver"""
def function(self, name: str) -> Callable:
"""Register Bedrock Agent function handler"""
def resolve(self, event: Dict, context: LambdaContext) -> BedrockFunctionResponse:
"""Resolve Bedrock Agent function event"""
@property
def current_event(self) -> BedrockAgentFunctionEvent:
"""Get current Bedrock Agent function event"""VPC Lattice Resolver
Resolver for Amazon VPC Lattice service-to-service communication.
class VPCLatticeResolver:
def __init__(
self,
cors: CORSConfig = None,
debug: bool = None,
serializer: Callable[[Dict], str] = None,
strip_prefixes: List[str] = None,
enable_validation: bool = False,
):
"""
Initialize VPC Lattice resolver.
Parameters:
- cors: CORS configuration
- debug: Enable debug mode
- serializer: Custom JSON serializer
- strip_prefixes: Path prefixes to strip
- enable_validation: Enable request validation
"""
# Same routing interface
def get(self, rule: str, **kwargs) -> Callable: ...
def post(self, rule: str, **kwargs) -> Callable: ...
def resolve(self, event: Dict, context: LambdaContext) -> Dict: ...
@property
def current_event(self) -> VPCLatticeEvent:
"""Get current VPC Lattice event"""
class VPCLatticeV2Resolver:
def __init__(
self,
cors: CORSConfig = None,
debug: bool = None,
serializer: Callable[[Dict], str] = None,
strip_prefixes: List[str] = None,
enable_validation: bool = False,
):
"""Initialize VPC Lattice V2 resolver"""
def resolve(self, event: Dict, context: LambdaContext) -> Dict: ...
@property
def current_event(self) -> VPCLatticeEventV2:
"""Get current VPC Lattice V2 event"""Response Classes
HTTP response objects for returning structured responses from event handlers.
class Response:
def __init__(
self,
status_code: int,
content_type: str = None,
body: str = None,
headers: Dict[str, str] = None,
cookies: List[str] = None,
compress: bool = None,
):
"""
Create HTTP response.
Parameters:
- status_code: HTTP status code
- content_type: Response content type
- body: Response body content
- headers: Additional HTTP headers
- cookies: Set-Cookie header values
- compress: Whether to compress response body
"""
class BedrockResponse:
def __init__(
self,
response: Dict[str, Any],
session_attributes: Dict[str, str] = None,
prompt_session_attributes: Dict[str, str] = None,
):
"""
Bedrock Agent response.
Parameters:
- response: Response data for agent
- session_attributes: Session attributes to persist
- prompt_session_attributes: Prompt session attributes
"""
class BedrockFunctionResponse:
def __init__(
self,
response: Dict[str, Any],
message_version: str = "1.0",
):
"""
Bedrock Agent function response.
Parameters:
- response: Function execution response
- message_version: Response message version
"""CORS Configuration
Cross-Origin Resource Sharing (CORS) configuration for HTTP APIs.
class CORSConfig:
def __init__(
self,
allow_origin: str = "*",
allow_headers: List[str] = None,
allow_methods: List[str] = None,
expose_headers: List[str] = None,
max_age: int = None,
allow_credentials: bool = False,
):
"""
Configure CORS settings.
Parameters:
- allow_origin: Allowed origins for cross-origin requests
- allow_headers: Allowed headers in preflight requests
- allow_methods: Allowed HTTP methods
- expose_headers: Headers exposed to client
- max_age: Cache duration for preflight requests (seconds)
- allow_credentials: Whether to allow credentials
"""Middleware Support
Base classes for creating custom middleware for request/response processing.
class BaseMiddlewareHandler:
def __init__(self):
"""Base middleware handler for event processing"""
def handler(
self,
event: Dict[str, Any],
context: LambdaContext,
next_middleware: "NextMiddleware",
) -> Dict[str, Any]:
"""
Process middleware logic.
Parameters:
- event: Incoming Lambda event
- context: Lambda runtime context
- next_middleware: Next middleware in chain
Returns:
Processed event or response
"""
class NextMiddleware:
def __init__(self, handler: Callable):
"""Next middleware wrapper"""
def __call__(
self,
event: Dict[str, Any],
context: LambdaContext,
) -> Dict[str, Any]:
"""Call next middleware or handler"""Usage Examples
REST API with Multiple Routes
from aws_lambda_powertools.event_handler import APIGatewayRestResolver
from aws_lambda_powertools.utilities.typing import LambdaContext
app = APIGatewayRestResolver()
@app.get("/users")
def list_users():
# Access request context
query_params = app.current_event.query_string_parameters or {}
# Return JSON response
return {
"users": [
{"id": 1, "name": "Alice"},
{"id": 2, "name": "Bob"}
],
"total": 2,
"limit": query_params.get("limit", 10)
}
@app.post("/users")
def create_user():
# Parse JSON body
user_data = app.current_event.json_body
# Validate required fields
if not user_data.get("name"):
return Response(
status_code=400,
body={"error": "Name is required"}
)
# Create user logic here
new_user = {"id": 3, "name": user_data["name"]}
return Response(
status_code=201,
body=new_user
)
@app.get("/users/<user_id>")
def get_user(user_id: str):
# Path parameters are automatically parsed
# Simulate user lookup
if user_id == "1":
return {"id": 1, "name": "Alice"}
else:
return Response(status_code=404, body={"error": "User not found"})
@app.exception_handler(ValueError)
def handle_validation_error(ex: ValueError):
return Response(
status_code=400,
body={"error": f"Validation failed: {str(ex)}"}
)
def lambda_handler(event: dict, context: LambdaContext) -> dict:
return app.resolve(event, context)GraphQL AppSync Resolver
from aws_lambda_powertools.event_handler import AppSyncResolver
from aws_lambda_powertools.utilities.typing import LambdaContext
app = AppSyncResolver()
@app.resolver(type_name="Query", field_name="getUser")
def get_user(user_id: str):
# Access GraphQL context
event = app.current_event
# Get arguments
user_id = event.arguments.get("id")
# Access identity info
user_identity = event.identity
# Return user data
return {
"id": user_id,
"name": f"User {user_id}",
"email": f"user{user_id}@example.com"
}
@app.resolver(type_name="Mutation", field_name="createUser")
def create_user():
event = app.current_event
user_input = event.arguments.get("input", {})
# Create user logic
new_user = {
"id": "new_123",
"name": user_input.get("name"),
"email": user_input.get("email")
}
return new_user
@app.batch_resolver(type_name="User", field_name="posts")
def get_user_posts(user_ids: List[str]):
# Batch resolver for efficient N+1 resolution
# This runs once for multiple User.posts requests
# Bulk fetch posts for all users
posts_by_user = fetch_posts_for_users(user_ids)
# Return posts in same order as user_ids
return [posts_by_user.get(user_id, []) for user_id in user_ids]
def lambda_handler(event: dict, context: LambdaContext):
return app.resolve(event, context)Bedrock Agent with Actions
from aws_lambda_powertools.event_handler import BedrockAgentResolver, BedrockResponse
from aws_lambda_powertools.utilities.typing import LambdaContext
app = BedrockAgentResolver()
@app.action("get_weather")
def get_weather():
event = app.current_event
# Extract parameters from agent
parameters = event.parameters
location = parameters.get("location", "unknown")
# Get weather data (mock)
weather_data = {
"location": location,
"temperature": "72°F",
"conditions": "sunny",
"forecast": "Clear skies expected"
}
return BedrockResponse(
response=weather_data,
session_attributes={"last_location": location}
)
@app.action("book_appointment")
def book_appointment():
event = app.current_event
parameters = event.parameters
date = parameters.get("date")
time = parameters.get("time")
service = parameters.get("service")
# Appointment booking logic
booking_result = {
"confirmation_id": "BOOK123",
"date": date,
"time": time,
"service": service,
"status": "confirmed"
}
return BedrockResponse(
response=booking_result,
session_attributes={
"last_booking": "BOOK123",
"user_preferences": {"service": service}
}
)
def lambda_handler(event: dict, context: LambdaContext):
return app.resolve(event, context)CORS and Error Handling
from aws_lambda_powertools.event_handler import (
APIGatewayRestResolver,
CORSConfig,
Response
)
from aws_lambda_powertools.utilities.typing import LambdaContext
# Configure CORS
cors_config = CORSConfig(
allow_origin="https://myapp.example.com",
allow_headers=["Content-Type", "Authorization"],
allow_methods=["GET", "POST", "PUT", "DELETE"],
expose_headers=["X-Custom-Header"],
max_age=3600,
allow_credentials=True
)
app = APIGatewayRestResolver(cors=cors_config)
@app.get("/api/data")
def get_data():
return {"message": "Hello from API"}
@app.exception_handler(ValueError)
def handle_validation_error(ex: ValueError):
return Response(
status_code=400,
body={"error": "Invalid input", "details": str(ex)},
headers={"X-Error-Type": "ValidationError"}
)
@app.exception_handler(KeyError)
def handle_missing_key(ex: KeyError):
return Response(
status_code=400,
body={"error": "Missing required field", "field": str(ex)}
)
@app.not_found()
def handle_not_found():
return Response(
status_code=404,
body={"error": "Resource not found"}
)
@app.unauthorized()
def handle_unauthorized():
return Response(
status_code=401,
body={"error": "Authentication required"},
headers={"WWW-Authenticate": "Bearer"}
)
def lambda_handler(event: dict, context: LambdaContext) -> dict:
return app.resolve(event, context)Types
from typing import Dict, List, Any, Union, Callable, Optional
from aws_lambda_powertools.utilities.data_classes import (
APIGatewayProxyEvent,
APIGatewayProxyEventV2,
ALBEvent,
AppSyncResolverEvent,
AppSyncResolverEventsEvent,
BedrockAgentEvent,
BedrockAgentFunctionEvent,
LambdaFunctionUrlEvent,
VPCLatticeEvent,
VPCLatticeEventV2,
)
# Route handler function signature
RouteHandler = Callable[..., Union[Dict[str, Any], Response, Any]]
# Exception handler function signature
ExceptionHandler = Callable[[Exception], Union[Dict[str, Any], Response]]
# Middleware function signature
MiddlewareHandler = Callable[
[Dict[str, Any], LambdaContext, NextMiddleware],
Dict[str, Any]
]
# Serializer function signature
Serializer = Callable[[Dict[str, Any]], str]Install with Tessl CLI
npx tessl i tessl/pypi-aws-lambda-powertoolsdocs
evals
scenario-1
scenario-2
scenario-3
scenario-4
scenario-5
scenario-6
scenario-7
scenario-8
scenario-9
scenario-10