Middleware

MODULE	DESCRIPTION
base
openapi_validation
schema_validation

CLASS	DESCRIPTION
BaseMiddlewareHandler	Base implementation for Middlewares to run code before and after in a chain.
NextMiddleware

BaseMiddlewareHandler

Bases: Generic[EventHandlerInstance], ABC

Base implementation for Middlewares to run code before and after in a chain.

This is the middleware handler function where middleware logic is implemented. The next middleware handler is represented by next_middleware, returning a Response object.

Example

Correlation ID Middleware

import requests

from aws_lambda_powertools import Logger
from aws_lambda_powertools.event_handler import APIGatewayRestResolver, Response
from aws_lambda_powertools.event_handler.middlewares import BaseMiddlewareHandler, NextMiddleware

app = APIGatewayRestResolver()
logger = Logger()


class CorrelationIdMiddleware(BaseMiddlewareHandler):
    def __init__(self, header: str):
        super().__init__()
        self.header = header

    def handler(self, app: APIGatewayRestResolver, next_middleware: NextMiddleware) -> Response:
        # BEFORE logic
        request_id = app.current_event.request_context.request_id
        correlation_id = app.current_event.headers.get(self.header, request_id)

        # Call next middleware or route handler ('/todos')
        response = next_middleware(app)

        # AFTER logic
        response.headers[self.header] = correlation_id

        return response


@app.get("/todos", middlewares=[CorrelationIdMiddleware(header="x-correlation-id")])
def get_todos():
    todos: requests.Response = requests.get("https://jsonplaceholder.typicode.com/todos")
    todos.raise_for_status()

    # for brevity, we'll limit to the first 10 only
    return {"todos": todos.json()[:10]}


@logger.inject_lambda_context
def lambda_handler(event, context):
    return app.resolve(event, context)

METHOD	DESCRIPTION
handler	The Middleware Handler

handler abstractmethod

handler(app: EventHandlerInstance, next_middleware: NextMiddleware) -> Response

The Middleware Handler

PARAMETER	DESCRIPTION
app	An instance of an Event Handler that implements ApiGatewayResolver TYPE: EventHandlerInstance
next_middleware	The next middleware handler in the chain TYPE: NextMiddleware

RETURNS	DESCRIPTION
Response	The response from the next middleware handler in the chain

Source code in aws_lambda_powertools/event_handler/middlewares/base.py

@abstractmethod
def handler(self, app: EventHandlerInstance, next_middleware: NextMiddleware) -> Response:
    """
    The Middleware Handler

    Parameters
    ----------
    app: EventHandlerInstance
        An instance of an Event Handler that implements ApiGatewayResolver
    next_middleware: NextMiddleware
        The next middleware handler in the chain

    Returns
    -------
    Response
        The response from the next middleware handler in the chain

    """
    raise NotImplementedError()

NextMiddleware

Bases: Protocol

base

CLASS	DESCRIPTION
BaseMiddlewareHandler	Base implementation for Middlewares to run code before and after in a chain.
NextMiddleware

BaseMiddlewareHandler

Bases: Generic[EventHandlerInstance], ABC

Base implementation for Middlewares to run code before and after in a chain.

This is the middleware handler function where middleware logic is implemented. The next middleware handler is represented by next_middleware, returning a Response object.

Example

Correlation ID Middleware

import requests

from aws_lambda_powertools import Logger
from aws_lambda_powertools.event_handler import APIGatewayRestResolver, Response
from aws_lambda_powertools.event_handler.middlewares import BaseMiddlewareHandler, NextMiddleware

app = APIGatewayRestResolver()
logger = Logger()


class CorrelationIdMiddleware(BaseMiddlewareHandler):
    def __init__(self, header: str):
        super().__init__()
        self.header = header

    def handler(self, app: APIGatewayRestResolver, next_middleware: NextMiddleware) -> Response:
        # BEFORE logic
        request_id = app.current_event.request_context.request_id
        correlation_id = app.current_event.headers.get(self.header, request_id)

        # Call next middleware or route handler ('/todos')
        response = next_middleware(app)

        # AFTER logic
        response.headers[self.header] = correlation_id

        return response


@app.get("/todos", middlewares=[CorrelationIdMiddleware(header="x-correlation-id")])
def get_todos():
    todos: requests.Response = requests.get("https://jsonplaceholder.typicode.com/todos")
    todos.raise_for_status()

    # for brevity, we'll limit to the first 10 only
    return {"todos": todos.json()[:10]}


@logger.inject_lambda_context
def lambda_handler(event, context):
    return app.resolve(event, context)

METHOD	DESCRIPTION
handler	The Middleware Handler

handler abstractmethod

handler(app: EventHandlerInstance, next_middleware: NextMiddleware) -> Response

The Middleware Handler

PARAMETER	DESCRIPTION
app	An instance of an Event Handler that implements ApiGatewayResolver TYPE: EventHandlerInstance
next_middleware	The next middleware handler in the chain TYPE: NextMiddleware

RETURNS	DESCRIPTION
Response	The response from the next middleware handler in the chain

Source code in aws_lambda_powertools/event_handler/middlewares/base.py

@abstractmethod
def handler(self, app: EventHandlerInstance, next_middleware: NextMiddleware) -> Response:
    """
    The Middleware Handler

    Parameters
    ----------
    app: EventHandlerInstance
        An instance of an Event Handler that implements ApiGatewayResolver
    next_middleware: NextMiddleware
        The next middleware handler in the chain

    Returns
    -------
    Response
        The response from the next middleware handler in the chain

    """
    raise NotImplementedError()

NextMiddleware

Bases: Protocol

openapi_validation

CLASS	DESCRIPTION
OpenAPIRequestValidationMiddleware	OpenAPI request validation middleware - validates only incoming requests.
OpenAPIResponseValidationMiddleware	OpenAPI response validation middleware - validates only outgoing responses.

OpenAPIRequestValidationMiddleware

OpenAPIRequestValidationMiddleware()

Bases: BaseMiddlewareHandler

OpenAPI request validation middleware - validates only incoming requests.

This middleware should be used first in the middleware chain to validate requests before they reach user middlewares.

Source code in aws_lambda_powertools/event_handler/middlewares/openapi_validation.py

def __init__(self):
    """Initialize the request validation middleware."""
    pass

OpenAPIResponseValidationMiddleware

OpenAPIResponseValidationMiddleware(validation_serializer: Callable[[Any], str] | None = None, has_response_validation_error: bool = False)

Bases: BaseMiddlewareHandler

OpenAPI response validation middleware - validates only outgoing responses.

This middleware should be used last in the middleware chain to validate responses only from route handlers, not from user middlewares.

PARAMETER	DESCRIPTION
validation_serializer	Optional serializer to use when serializing the response for validation. Use it when you have a custom type that cannot be serialized by the default jsonable_encoder. TYPE: Callable DEFAULT: None
has_response_validation_error	Optional flag used to distinguish between payload and validation errors. By setting this flag to True, ResponseValidationError will be raised if response could not be validated. TYPE: bool DEFAULT: False

Source code in aws_lambda_powertools/event_handler/middlewares/openapi_validation.py

def __init__(
    self,
    validation_serializer: Callable[[Any], str] | None = None,
    has_response_validation_error: bool = False,
):
    """
    Initialize the response validation middleware.

    Parameters
    ----------
    validation_serializer : Callable, optional
        Optional serializer to use when serializing the response for validation.
        Use it when you have a custom type that cannot be serialized by the default jsonable_encoder.

    has_response_validation_error: bool, optional
        Optional flag used to distinguish between payload and validation errors.
        By setting this flag to True, ResponseValidationError will be raised if response could not be validated.
    """
    self._validation_serializer = validation_serializer
    self._has_response_validation_error = has_response_validation_error

schema_validation

CLASS	DESCRIPTION
SchemaValidationMiddleware	Middleware to validate API request and response against JSON Schema using the Validation utility.

SchemaValidationMiddleware

SchemaValidationMiddleware(inbound_schema: dict, inbound_formats: dict | None = None, outbound_schema: dict | None = None, outbound_formats: dict | None = None)

Bases: BaseMiddlewareHandler

Middleware to validate API request and response against JSON Schema using the Validation utility.

Example

Validating incoming event

import requests

from aws_lambda_powertools import Logger
from aws_lambda_powertools.event_handler import APIGatewayRestResolver, Response
from aws_lambda_powertools.event_handler.middlewares import BaseMiddlewareHandler, NextMiddleware
from aws_lambda_powertools.event_handler.middlewares.schema_validation import SchemaValidationMiddleware

app = APIGatewayRestResolver()
logger = Logger()
json_schema_validation = SchemaValidationMiddleware(inbound_schema=INCOMING_JSON_SCHEMA)


@app.get("/todos", middlewares=[json_schema_validation])
def get_todos():
    todos: requests.Response = requests.get("https://jsonplaceholder.typicode.com/todos")
    todos.raise_for_status()

    # for brevity, we'll limit to the first 10 only
    return {"todos": todos.json()[:10]}


@logger.inject_lambda_context
def lambda_handler(event, context):
    return app.resolve(event, context)

PARAMETER	DESCRIPTION
inbound_schema	JSON Schema to validate incoming event TYPE: dict
inbound_formats	Custom formats containing a key (e.g. int64) and a value expressed as regex or callback returning bool, by default None JSON Schema to validate outbound event, by default None TYPE: dict | None DEFAULT: None
outbound_formats	Custom formats containing a key (e.g. int64) and a value expressed as regex or callback returning bool, by default None TYPE: dict | None DEFAULT: None

METHOD	DESCRIPTION
handler	Validates incoming JSON payload (body) against JSON Schema provided.

Source code in aws_lambda_powertools/event_handler/middlewares/schema_validation.py

def __init__(
    self,
    inbound_schema: dict,
    inbound_formats: dict | None = None,
    outbound_schema: dict | None = None,
    outbound_formats: dict | None = None,
):
    """See [Validation utility](https://docs.powertools.aws.dev/lambda/python/latest/utilities/validation/) docs for examples on all parameters.

    Parameters
    ----------
    inbound_schema : dict
        JSON Schema to validate incoming event
    inbound_formats : dict | None, optional
        Custom formats containing a key (e.g. int64) and a value expressed as regex or callback returning bool, by default None
        JSON Schema to validate outbound event, by default None
    outbound_formats : dict | None, optional
        Custom formats containing a key (e.g. int64) and a value expressed as regex or callback returning bool, by default None
    """  # noqa: E501
    super().__init__()
    self.inbound_schema = inbound_schema
    self.inbound_formats = inbound_formats
    self.outbound_schema = outbound_schema
    self.outbound_formats = outbound_formats

handler

handler(app: EventHandlerInstance, next_middleware: NextMiddleware) -> Response

Validates incoming JSON payload (body) against JSON Schema provided.

PARAMETER	DESCRIPTION
app	An instance of an Event Handler TYPE: EventHandlerInstance
next_middleware	Callable to get response from the next middleware or route handler in the chain TYPE: NextMiddleware

RETURNS	DESCRIPTION
Response	It can return three types of response objects Original response: Propagates HTTP response returned from the next middleware if validation succeeds HTTP 400: Payload or response failed JSON Schema validation HTTP 500: JSON Schema provided has incorrect format

Source code in aws_lambda_powertools/event_handler/middlewares/schema_validation.py

def handler(self, app: EventHandlerInstance, next_middleware: NextMiddleware) -> Response:
    """Validates incoming JSON payload (body) against JSON Schema provided.

    Parameters
    ----------
    app : EventHandlerInstance
        An instance of an Event Handler
    next_middleware : NextMiddleware
        Callable to get response from the next middleware or route handler in the chain

    Returns
    -------
    Response
        It can return three types of response objects

        - Original response: Propagates HTTP response returned from the next middleware if validation succeeds
        - HTTP 400: Payload or response failed JSON Schema validation
        - HTTP 500: JSON Schema provided has incorrect format
    """
    try:
        validate(event=app.current_event.json_body, schema=self.inbound_schema, formats=self.inbound_formats)
    except SchemaValidationError as error:
        return self.bad_request(error)
    except InvalidSchemaFormatError as error:
        return self.bad_config(error)

    result = next_middleware(app)

    if self.outbound_formats is not None:
        try:
            validate(event=result.body, schema=self.inbound_schema, formats=self.inbound_formats)
        except SchemaValidationError as error:
            return self.bad_response(error)
        except InvalidSchemaFormatError as error:
            return self.bad_config(error)

    return result