CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-zappa

Server-less Python Web Services for AWS Lambda and API Gateway

Overview
Eval results
Files

wsgi-processing.mddocs/

WSGI Request Processing

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.

Capabilities

Event to WSGI Conversion

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
    """

Payload Processing

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)
    """

Response Formatting

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

ZappaWSGIMiddleware

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
        """

Utility Functions

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
    """

Constants

# HTTP methods that may contain binary data
BINARY_METHODS = ['POST', 'PUT', 'PATCH']

Usage Examples

Basic WSGI Conversion

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 response

Django Integration

from 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)

Flask Integration

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)

Custom Response Processing

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 response

Binary Content Handling

from 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

docs

async-tasks.md

cli-operations.md

core-deployment.md

index.md

package-utilities.md

request-processing.md

ssl-management.md

utilities.md

wsgi-processing.md

tile.json