AWS Lambda Adapter

AwsApiGatewayAdapter

AwsApiGatewayAdapter(app: RestApplication, metrics_publisher: Union[MetricsPublisher, None, _DefaultPublisher] = _DEFAULT_PUBLISHER, enable_metrics: Optional[bool] = None, namespace: Optional[str] = None, service_name: Optional[str] = None, metrics_resolution: int = 60)

Bases: Adapter

Adapter for AWS API Gateway Lambda proxy integration events.

This adapter handles events from: - API Gateway REST APIs (v1) - payload format 1.0 - API Gateway HTTP APIs (v2) - payload format 2.0 - Application Load Balancer (ALB) - Lambda Function URLs (v2 format)

Version detection is automatic based on event structure: - v1 events use httpMethod at top level - v2 events have version: "2.0" field - ALB events have requestContext.elb field

Follows ASGI patterns for header and parameter handling: - Headers are normalized to lowercase for case-insensitive matching - Query parameters are parsed consistently - Body encoding is handled transparently

Initialize the adapter with a RestApplication instance.

Automatically executes startup handlers during cold start and configures CloudWatch EMF metrics unless disabled.

Parameters:

Name	Type	Description	Default
app	RestApplication	The RestApplication instance to execute requests against	required
metrics_publisher	Union[MetricsPublisher, None, _DefaultPublisher]	Metrics publisher. Defaults to CloudWatchEMFPublisher. Pass None to explicitly disable metrics.	_DEFAULT_PUBLISHER
enable_metrics	Optional[bool]	Explicitly enable/disable metrics.	None
namespace	Optional[str]	CloudWatch namespace (overrides env var)	None
service_name	Optional[str]	Service name for dimension (overrides env var)	None
metrics_resolution	int	Default resolution, 1 or 60 seconds (default: 60)	60

Examples:

Auto EMF with custom namespace

adapter = AwsApiGatewayAdapter( app, namespace="MyApp/API", service_name="user-service" )

High-resolution metrics

adapter = AwsApiGatewayAdapter( app, namespace="MyApp/API", metrics_resolution=1 )

Disable metrics

adapter = AwsApiGatewayAdapter(app, enable_metrics=False)

Source code in packages/restmachine-aws/src/restmachine_aws/adapter.py

def __init__(self,
             app: RestApplication,
             metrics_publisher: Union[MetricsPublisher, None, _DefaultPublisher] = _DEFAULT_PUBLISHER,
             enable_metrics: Optional[bool] = None,
             namespace: Optional[str] = None,
             service_name: Optional[str] = None,
             metrics_resolution: int = 60):
    """
    Initialize the adapter with a RestApplication instance.

    Automatically executes startup handlers during cold start and
    configures CloudWatch EMF metrics unless disabled.

    Args:
        app: The RestApplication instance to execute requests against
        metrics_publisher: Metrics publisher. Defaults to CloudWatchEMFPublisher.
                          Pass None to explicitly disable metrics.
        enable_metrics: Explicitly enable/disable metrics.
        namespace: CloudWatch namespace (overrides env var)
        service_name: Service name for dimension (overrides env var)
        metrics_resolution: Default resolution, 1 or 60 seconds (default: 60)

    Examples:
        # Auto EMF with custom namespace
        adapter = AwsApiGatewayAdapter(
            app,
            namespace="MyApp/API",
            service_name="user-service"
        )

        # High-resolution metrics
        adapter = AwsApiGatewayAdapter(
            app,
            namespace="MyApp/API",
            metrics_resolution=1
        )

        # Disable metrics
        adapter = AwsApiGatewayAdapter(app, enable_metrics=False)
    """
    self.app = app

    # Determine if metrics should be enabled
    metrics_enabled = self._should_enable_metrics(enable_metrics)

    # Auto-configure publisher if not provided
    publisher: Optional[MetricsPublisher]
    if isinstance(metrics_publisher, _DefaultPublisher):
        if metrics_enabled:
            publisher = self._create_default_publisher(
                namespace=namespace,
                service_name=service_name,
                resolution=metrics_resolution
            )
            self._configure_default_logging()
        else:
            publisher = None
    else:
        publisher = metrics_publisher

    self.metrics_handler = MetricsHandler(app, publisher)

    # Execute startup handlers during Lambda cold start
    # This ensures database connections, API clients, etc. are initialized
    # before the first request is processed
    if hasattr(app, '_startup_handlers') and app._startup_handlers:
        app.startup_sync()

Functions

handle_event

handle_event(event: Dict[str, Any], context: Optional[Any] = None) -> Dict[str, Any]

Handle an AWS API Gateway event with metrics support.

Parameters:

Name	Type	Description	Default
event	Dict[str, Any]	AWS API Gateway event dictionary	required
context	Optional[Any]	AWS Lambda context (optional)	None

Returns:

Type	Description
Dict[str, Any]	AWS API Gateway response dictionary

Source code in packages/restmachine-aws/src/restmachine_aws/adapter.py

def handle_event(self, event: Dict[str, Any], context: Optional[Any] = None) -> Dict[str, Any]:
    """
    Handle an AWS API Gateway event with metrics support.

    Args:
        event: AWS API Gateway event dictionary
        context: AWS Lambda context (optional)

    Returns:
        AWS API Gateway response dictionary
    """
    return cast(Dict[str, Any], self.metrics_handler.handle_request(
        event,
        context,
        convert_fn=self.convert_to_request,
        execute_fn=self.app.execute,
        response_fn=self.convert_from_response
    ))

convert_to_request

convert_to_request(event: Dict[str, Any], context: Optional[Any] = None) -> Request

Convert AWS event to Request object.

Automatically detects and handles: - API Gateway (v1, v2) events - Application Load Balancer (ALB) events - Lambda Function URL events

Parameters:

Name	Type	Description	Default
event	Dict[str, Any]	AWS event dictionary	required
context	Optional[Any]	AWS Lambda context (optional)	None

Returns:

Type	Description
Request	Request object

Source code in packages/restmachine-aws/src/restmachine_aws/adapter.py

def convert_to_request(self, event: Dict[str, Any], context: Optional[Any] = None) -> Request:
    """
    Convert AWS event to Request object.

    Automatically detects and handles:
    - API Gateway (v1, v2) events
    - Application Load Balancer (ALB) events
    - Lambda Function URL events

    Args:
        event: AWS event dictionary
        context: AWS Lambda context (optional)

    Returns:
        Request object
    """
    # Detect event type and delegate to appropriate parser
    if self._is_alb_event(event):
        return self._parse_alb_event(event, context)
    else:
        return self._parse_apigw_event(event, context)

convert_from_response

convert_from_response(response: Response, event: Dict[str, Any], context: Optional[Any] = None) -> Dict[str, Any]

Convert Response object to AWS API Gateway response format.

Handles proper JSON serialization, header encoding, streaming bodies, and file paths. For streaming bodies and Path objects, reads the entire content since Lambda requires complete responses.

Range Request Support: - Detects 206 Partial Content responses (response.is_range_response()) - Extracts only the requested byte range from Path, stream, or bytes body - Base64 encodes the range content for transmission

Parameters:

Name	Type	Description	Default
response	Response	Response from the app	required
event	Dict[str, Any]	Original AWS API Gateway event	required
context	Optional[Any]	AWS Lambda context (optional)	None

Returns:

Type	Description
Dict[str, Any]	AWS API Gateway response dictionary

Source code in packages/restmachine-aws/src/restmachine_aws/adapter.py

def convert_from_response(self, response: Response, event: Dict[str, Any], context: Optional[Any] = None) -> Dict[str, Any]:
    """
    Convert Response object to AWS API Gateway response format.

    Handles proper JSON serialization, header encoding, streaming bodies, and file paths.
    For streaming bodies and Path objects, reads the entire content since Lambda requires complete responses.

    Range Request Support:
    - Detects 206 Partial Content responses (response.is_range_response())
    - Extracts only the requested byte range from Path, stream, or bytes body
    - Base64 encodes the range content for transmission

    Args:
        response: Response from the app
        event: Original AWS API Gateway event
        context: AWS Lambda context (optional)

    Returns:
        AWS API Gateway response dictionary
    """
    # Handle range responses specially
    if response.is_range_response():
        return self._convert_range_response(response)

    # Convert body to string and track if we used base64 encoding
    is_base64 = False

    if response.body is None:
        body_str = ""
    elif isinstance(response.body, Path):
        # Path object - read the file since Lambda requires complete response
        path_obj = response.body
        if path_obj.exists() and path_obj.is_file():
            with path_obj.open('rb') as f:
                body_bytes = f.read()
            try:
                body_str = body_bytes.decode('utf-8')
            except UnicodeDecodeError:
                # If not valid UTF-8, return as base64
                import base64
                body_str = base64.b64encode(body_bytes).decode('ascii')
                is_base64 = True
        else:
            # File doesn't exist
            body_str = ""
    elif isinstance(response.body, io.IOBase):
        # Streaming body - read the entire stream since Lambda requires complete response
        body_bytes = response.body.read()
        try:
            body_str = body_bytes.decode('utf-8')
        except (UnicodeDecodeError, AttributeError):
            # If not valid UTF-8, return as base64
            import base64
            body_str = base64.b64encode(body_bytes).decode('ascii')
            is_base64 = True
    elif isinstance(response.body, bytes):
        # Raw bytes - try to decode as UTF-8, otherwise base64
        try:
            body_str = response.body.decode('utf-8')
        except UnicodeDecodeError:
            import base64
            body_str = base64.b64encode(response.body).decode('ascii')
            is_base64 = True
    elif isinstance(response.body, (dict, list)):
        body_str = json.dumps(response.body)
    elif isinstance(response.body, (str, int, float, bool)):
        body_str = str(response.body)
    else:
        body_str = str(response.body)

    # Build the API Gateway response
    # Convert MultiValueHeaders to dict (first value for each header)
    if response.headers:
        if isinstance(response.headers, MultiValueHeaders):
            headers_dict = response.headers.to_dict()
        else:
            headers_dict = dict(response.headers)
    else:
        headers_dict = {}

    # Ensure Content-Type is set for JSON responses
    if isinstance(response.body, (dict, list)) and "content-type" not in (
        {k.lower(): k for k in headers_dict.keys()}
    ):
        headers_dict["Content-Type"] = "application/json"

    # Build and return the API Gateway response
    api_response = {
        "statusCode": response.status_code,
        "headers": headers_dict,
        "body": body_str,
        "isBase64Encoded": is_base64
    }

    return api_response

Overview

The AwsApiGatewayAdapter converts AWS Lambda events into RestMachine requests and responses back into Lambda-compatible formats. It automatically detects and handles multiple event sources.

Supported Event Sources

The adapter automatically detects and handles:

• API Gateway REST API (v1) - Payload format 1.0
• API Gateway HTTP API (v2) - Payload format 2.0
• Application Load Balancer (ALB) - ALB target integration
• Lambda Function URLs - Direct HTTPS endpoints (uses v2 format)

Version detection is automatic based on the event structure.

Basic Usage

from restmachine import RestApplication
from restmachine_aws import AwsApiGatewayAdapter

app = RestApplication()

@app.get("/hello/{name}")
def hello(request):
    name = request.path_params['name']
    return {"message": f"Hello, {name}!"}

# Create adapter
adapter = AwsApiGatewayAdapter(app)

def lambda_handler(event, context):
    """AWS Lambda handler function."""
    return adapter.handle_event(event, context)

Automatic Startup

The adapter automatically runs startup handlers during Lambda cold start:

@app.on_startup
def database():
    """Runs once per Lambda container."""
    print("Opening database connection...")
    return create_db_connection()

@app.get("/users/{user_id}")
def get_user(database, request):
    # Database connection is already initialized
    user_id = request.path_params['user_id']
    return database.get_user(user_id)

Startup handlers execute when the Lambda container initializes, not on every request. This ensures expensive operations (like opening database connections) only happen during cold starts.

Event Format Detection

API Gateway v1 (REST API)

{
    "httpMethod": "GET",
    "path": "/users/123",
    "headers": {"Accept": "application/json"},
    "pathParameters": {"user_id": "123"},
    "queryStringParameters": {"page": "1"},
    "body": null,
    "requestContext": {
        "requestId": "abc123",
        "authorizer": {...}
    }
}

API Gateway v2 (HTTP API)

{
    "version": "2.0",
    "routeKey": "GET /users/{user_id}",
    "rawPath": "/users/123",
    "headers": {"accept": "application/json"},
    "pathParameters": {"user_id": "123"},
    "queryStringParameters": {"page": "1"},
    "body": null,
    "requestContext": {
        "http": {
            "method": "GET",
            "path": "/users/123"
        }
    }
}

Application Load Balancer (ALB)

{
    "httpMethod": "GET",
    "path": "/users/123",
    "headers": {"accept": "application/json"},
    "queryStringParameters": {"page": "1"},
    "body": null,
    "requestContext": {
        "elb": {
            "targetGroupArn": "arn:aws:..."
        }
    }
}

The adapter automatically detects the format and extracts the correct fields.

Header Handling

Headers are normalized to lowercase for case-insensitive matching, following ASGI conventions:

# API Gateway event
{
    "headers": {
        "Content-Type": "application/json",
        "X-Custom-Header": "value"
    }
}

# Converted to
request.headers = {
    "content-type": "application/json",
    "x-custom-header": "value"
}

Access headers in your handlers:

@app.get("/api/data")
def get_data(request):
    content_type = request.headers.get("content-type")
    custom = request.headers.get("x-custom-header")
    return {"content_type": content_type, "custom": custom}

Multi-Value Headers

The adapter properly handles multi-value headers (cookies, set-cookie, etc.):

# API Gateway v1 with multiValueHeaders
{
    "multiValueHeaders": {
        "cookie": ["session=abc123", "tracking=xyz789"]
    }
}

# Converted to RestMachine request
request.headers = MultiValueHeaders({
    "cookie": ["session=abc123", "tracking=xyz789"]
})

Query Parameters

Query parameters are parsed and available as dictionaries:

# Event
{
    "queryStringParameters": {
        "page": "2",
        "limit": "10"
    }
}

# Access in handler
@app.get("/users")
def list_users(request):
    page = int(request.query_params.get("page", 1))
    limit = int(request.query_params.get("limit", 20))
    return {"page": page, "limit": limit}

Request Body

The adapter handles both plain text and base64-encoded bodies:

# Plain text body
{
    "body": '{"name": "Alice", "email": "alice@example.com"}',
    "isBase64Encoded": false
}

# Base64-encoded body (binary data)
{
    "body": "iVBORw0KGgoAAAANSUhEUgAA...",
    "isBase64Encoded": true
}

# Access in handler
@app.post("/users")
def create_user(request):
    import json
    data = json.loads(request.body)
    return {"created": data}, 201

Response Conversion

RestMachine responses are automatically converted to Lambda-compatible formats:

# RestMachine response
Response(
    status_code=200,
    body='{"message": "Success"}',
    headers={"Content-Type": "application/json"}
)

# Converted to API Gateway v1 format
{
    "statusCode": 200,
    "body": '{"message": "Success"}',
    "headers": {"Content-Type": "application/json"},
    "multiValueHeaders": {},
    "isBase64Encoded": false
}

# Converted to API Gateway v2 format
{
    "statusCode": 200,
    "body": '{"message": "Success"}',
    "headers": {"content-type": "application/json"},
    "cookies": [],
    "isBase64Encoded": false
}

Error Handling

The adapter handles errors gracefully:

@app.get("/users/{user_id}")
def get_user(request):
    user_id = request.path_params['user_id']
    if not user_exists(user_id):
        return {"error": "User not found"}, 404
    return {"id": user_id, "name": "Alice"}

# Returns proper Lambda error response
{
    "statusCode": 404,
    "body": '{"error": "User not found"}',
    "headers": {"Content-Type": "application/json"}
}

Method Reference

handle_event(event, context)

Main entry point for Lambda handler.

Parameters: - event (dict): AWS Lambda event dictionary - context (optional): AWS Lambda context object

Returns: - dict: Lambda-compatible response dictionary

Example:

def lambda_handler(event, context):
    return adapter.handle_event(event, context)

convert_to_request(event, context)

Convert AWS event to RestMachine Request object.

Parameters: - event (dict): AWS Lambda event - context (optional): Lambda context

Returns: - Request: RestMachine Request object

convert_from_response(response, event, context)

Convert RestMachine Response to Lambda response format.

Parameters: - response (Response): RestMachine response - event (dict): Original AWS event (for format detection) - context (optional): Lambda context

Returns: - dict: Lambda-compatible response

See Also

• Lambda Deployment Guide - Complete deployment guide
• Lambda Extensions - Shutdown handler support
• Lifecycle Handlers - Startup and shutdown patterns