Server-less Python Web Services for AWS Lambda and API Gateway
Lambda function handlers and WSGI request processing for converting AWS Lambda events into standard HTTP requests that can be processed by web applications like Django and Flask. This capability enables seamless deployment of existing web applications to serverless infrastructure.
Main entry point for AWS Lambda function execution and request routing.
def lambda_handler(event, context):
"""
Global Lambda handler entry point for AWS Lambda runtime.
Processes incoming Lambda events and routes them to appropriate
handlers based on event type and source.
Parameters:
- event: dict, AWS Lambda event data
- context: LambdaContext, AWS Lambda runtime context
Returns:
dict: HTTP response with statusCode, headers, and body
"""class LambdaHandler:
@classmethod
def lambda_handler(cls, event, context):
"""
Class method entry point for Lambda event processing.
Manages singleton instance and delegates to handler method.
Parameters:
- event: dict, AWS Lambda event data
- context: LambdaContext, AWS Lambda runtime context
Returns:
dict: HTTP response
"""Core request processing and WSGI application execution.
def handler(self, event, context):
"""
Process Lambda events and route to WSGI application.
Converts Lambda events to WSGI environ dict, executes application,
and formats response for API Gateway.
Parameters:
- event: dict, Lambda event from API Gateway
- context: LambdaContext, Lambda runtime context
Returns:
dict: API Gateway response format
"""def run_function(self, event, context):
"""
Execute application function with Lambda event and context.
Loads and executes the target application function with
Lambda event data and runtime context.
Parameters:
- event: dict, Lambda event data
- context: LambdaContext, Lambda runtime context
Returns:
any: Function return value
"""Import and load application modules and functions at runtime.
@staticmethod
def import_module_and_get_function(whole_function):
"""
Import module and return function from dotted path.
Dynamically imports module and retrieves function for execution.
Parameters:
- whole_function: str, dotted path to function (module.function)
Returns:
callable: The imported function
"""Load application code and settings from remote sources.
def load_remote_project_archive(self, project_s3_bucket, project_s3_key):
"""
Load project code from remote ZIP archive in S3.
Downloads and extracts project archive to Lambda runtime
for execution of remotely stored applications.
Parameters:
- project_s3_bucket: str, S3 bucket containing project archive
- project_s3_key: str, S3 key for project ZIP file
Returns:
bool: True if loading successful
"""def load_remote_settings(self, settings_s3_bucket, settings_s3_key):
"""
Load Zappa settings from remote S3 configuration file.
Downloads settings file from S3 and applies configuration
to current Lambda execution environment.
Parameters:
- settings_s3_bucket: str, S3 bucket containing settings
- settings_s3_key: str, S3 key for settings file
Returns:
dict: Loaded settings configuration
"""Handle keep-warm ping events to prevent Lambda cold starts.
def keep_warm_callback(event, context):
"""
Handle keep-warm ping events for Lambda function.
Responds to scheduled keep-warm events to maintain Lambda
function in warm state and reduce cold start latency.
Parameters:
- event: dict, CloudWatch Events scheduled event
- context: LambdaContext, Lambda runtime context
Returns:
dict: Keep-warm response
"""Convert Lambda events to WSGI requests and process responses.
def create_wsgi_request(event, script_name=None, base_path=None, trailing_slash=True):
"""
Create WSGI environ dictionary from Lambda event.
Converts API Gateway Lambda proxy integration event into
standard WSGI environ dict for web application processing.
Parameters:
- event: dict, API Gateway Lambda proxy event
- script_name: str, WSGI SCRIPT_NAME value
- base_path: str, base path for the application
- trailing_slash: bool, add trailing slash to paths
Returns:
dict: WSGI environ dictionary
"""def process_lambda_payload_v1(event):
"""
Process API Gateway v1.0 payload format.
Extracts HTTP request data from API Gateway v1.0 Lambda
proxy integration event format.
Parameters:
- event: dict, API Gateway v1.0 event
Returns:
tuple: (method, path, headers, body, query_params)
"""def process_lambda_payload_v2(event):
"""
Process API Gateway v2.0 payload format.
Extracts HTTP request data from API Gateway v2.0 Lambda
proxy integration event format.
Parameters:
- event: dict, API Gateway v2.0 event
Returns:
tuple: (method, path, headers, body, query_params)
"""Format responses for API Gateway integration.
def common_log(environ, response, response_time=None):
"""
Create Apache Common Log Format entries.
Generates standardized log entries from WSGI environ
and response data for request monitoring.
Parameters:
- environ: dict, WSGI environ dictionary
- response: dict, HTTP response data
- response_time: float, request processing time in seconds
Returns:
str: Common Log Format entry
"""def get_wsgi_string(wsgi_string):
"""
Encode string for WSGI compatibility.
Ensures proper string encoding for WSGI environ values
and HTTP headers.
Parameters:
- wsgi_string: str, input string to encode
Returns:
str: WSGI-compatible encoded string
"""class ZappaWSGIMiddleware:
"""
WSGI middleware for Zappa-specific request handling.
Provides WSGI application interface with Zappa-specific
processing for Lambda/API Gateway integration.
"""
def __init__(self, application):
"""
Initialize middleware with WSGI application.
Parameters:
- application: callable, WSGI application
"""
def __call__(self, environ, start_response):
"""
WSGI application interface.
Processes WSGI requests with Zappa-specific handling
and delegates to wrapped application.
Parameters:
- environ: dict, WSGI environ dictionary
- start_response: callable, WSGI start_response function
Returns:
iterable: WSGI response
"""# HTTP methods that may contain binary data
BINARY_METHODS = ['POST', 'PUT', 'PATCH']from zappa.handler import lambda_handler
# Use as Lambda function entry point
def my_lambda_handler(event, context):
return lambda_handler(event, context)from zappa.handler import LambdaHandler
class CustomHandler(LambdaHandler):
def handler(self, event, context):
# Custom preprocessing
processed_event = self.preprocess_event(event)
# Call parent handler
response = super().handler(processed_event, context)
# Custom postprocessing
return self.postprocess_response(response)
# Use custom handler
handler_instance = CustomHandler()
lambda_handler = handler_instance.lambda_handlerfrom zappa.wsgi import create_wsgi_request
from zappa.middleware import ZappaWSGIMiddleware
# Wrap existing WSGI app
from myapp import wsgi_application
zappa_app = ZappaWSGIMiddleware(wsgi_application)
def lambda_handler(event, context):
# Convert Lambda event to WSGI
environ = create_wsgi_request(event)
# Process through WSGI app
response = zappa_app(environ, start_response)
return responseInstall with Tessl CLI
npx tessl i tessl/pypi-zappa