Server-less Python Web Services for AWS Lambda and API Gateway
WSGI utilities for converting Lambda events to standard HTTP requests and middleware for AWS-specific processing requirements. These functions enable seamless integration of existing WSGI applications with AWS Lambda and API Gateway.
Convert AWS Lambda events to WSGI environ dictionaries for processing by web applications.
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
"""Process different API Gateway payload formats.
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
"""WSGI middleware providing Zappa-specific request handling and AWS compatibility.
class ZappaWSGIMiddleware:
"""
WSGI middleware for Zappa-specific request handling.
Provides WSGI application interface with Zappa-specific
processing for Lambda/API Gateway integration including
header manipulation for AWS compatibility.
"""
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
"""def all_casings(input_text):
"""
Generate all case permutations of a string.
Used for case-insensitive header matching in AWS
Lambda/API Gateway environments.
Parameters:
- input_text: str, text to generate casings for
Returns:
list: All possible case permutations
"""# HTTP methods that may contain binary data
BINARY_METHODS = ['POST', 'PUT', 'PATCH']from zappa.wsgi import create_wsgi_request
def lambda_handler(event, context):
# Convert Lambda event to WSGI environ
environ = create_wsgi_request(event)
# Use with existing WSGI application
response = my_wsgi_app(environ, start_response)
return responsefrom zappa.wsgi import create_wsgi_request, ZappaWSGIMiddleware
from django.core.wsgi import get_wsgi_application
# Wrap Django WSGI app with Zappa middleware
django_app = get_wsgi_application()
zappa_app = ZappaWSGIMiddleware(django_app)
def lambda_handler(event, context):
environ = create_wsgi_request(event)
return zappa_app(environ, start_response)from zappa.wsgi import create_wsgi_request, ZappaWSGIMiddleware
from flask import Flask
app = Flask(__name__)
zappa_app = ZappaWSGIMiddleware(app)
@app.route('/')
def hello():
return 'Hello from Lambda!'
def lambda_handler(event, context):
environ = create_wsgi_request(event)
return zappa_app(environ, start_response)from zappa.wsgi import common_log, process_lambda_payload_v1
import time
def lambda_handler(event, context):
start_time = time.time()
# Process API Gateway payload
method, path, headers, body, query_params = process_lambda_payload_v1(event)
# Create WSGI environ
environ = create_wsgi_request(event)
# Process request
response = my_app(environ, start_response)
# Log request
response_time = time.time() - start_time
log_entry = common_log(environ, response, response_time)
print(log_entry)
return responsefrom zappa.wsgi import BINARY_METHODS, get_wsgi_string
import base64
def lambda_handler(event, context):
environ = create_wsgi_request(event)
# Handle binary content for specific methods
if environ['REQUEST_METHOD'] in BINARY_METHODS:
if event.get('isBase64Encoded', False):
body = base64.b64decode(event['body'])
environ['wsgi.input'] = io.BytesIO(body)
return my_app(environ, start_response)Install with Tessl CLI
npx tessl i tessl/pypi-zappa