Skip to content

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
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 
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
@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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
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 
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
@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
OpenAPIValidationMiddleware

OpenAPIValidationMiddleware is a middleware that validates the request against the OpenAPI schema defined by the

OpenAPIValidationMiddleware 

OpenAPIValidationMiddleware(
    validation_serializer: (
        Callable[[Any], str] | None
    ) = None,
)

Bases: BaseMiddlewareHandler

OpenAPIValidationMiddleware is a middleware that validates the request against the OpenAPI schema defined by the Lambda handler. It also validates the response against the OpenAPI schema defined by the Lambda handler. It should not be used directly, but rather through the enable_validation parameter of the ApiGatewayResolver.

Example 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
from pydantic import BaseModel

from aws_lambda_powertools.event_handler.api_gateway import (
    APIGatewayRestResolver,
)

class Todo(BaseModel):
  name: str

app = APIGatewayRestResolver(enable_validation=True)

@app.get("/todos")
def get_todos(): list[Todo]:
  return [Todo(name="hello world")]
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

Source code in aws_lambda_powertools/event_handler/middlewares/openapi_validation.py 
61
62
63
64
65
66
67
68
69
70
71
def __init__(self, validation_serializer: Callable[[Any], str] | None = None):
    """
    Initialize the OpenAPIValidationMiddleware.

    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.
    """
    self._validation_serializer = validation_serializer

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
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 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
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 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
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