Middleware factory provides a decorator factory to create your own middleware to run logic before, and after each Lambda invocation synchronously.

## Key features

- Run logic before, after, and handle exceptions
- Built-in tracing opt-in capability

## Getting started

> [!TIP]
>
> All examples shared in this documentation are available within the [project repository](<https://github.com/aws-powertools/powertools-lambda-python/tree/develop/examples>).

You might need a custom middleware to abstract non-functional code. These are often custom authorization or any reusable logic you might need to run before/after a Lambda function invocation.

### Middleware with no params

You can create your own middleware using `lambda_handler_decorator`. The decorator factory expects 3 arguments in your function signature:

- **handler** - Lambda function handler
- **event** - Lambda function invocation event
- **context** - Lambda function context object

### Middleware with before logic

```
from collections.abc import Callable
from dataclasses import dataclass, field
from uuid import uuid4

from aws_lambda_powertools.middleware_factory import lambda_handler_decorator
from aws_lambda_powertools.utilities.jmespath_utils import (
    envelopes,
    query,
)
from aws_lambda_powertools.utilities.typing import LambdaContext


@dataclass
class Payment:
    user_id: str
    order_id: str
    amount: float
    status_id: str
    payment_id: str = field(default_factory=lambda: f"{uuid4()}")


class PaymentError(Exception): ...


@lambda_handler_decorator
def middleware_before(
    handler: Callable[[dict, LambdaContext], dict],
    event: dict,
    context: LambdaContext,
) -> dict:
    # extract payload from a EventBridge event
    detail: dict = query(data=event, envelope=envelopes.EVENTBRIDGE)

    # check if status_id exists in payload, otherwise add default state before processing payment
    if "status_id" not in detail:
        event["detail"]["status_id"] = "pending"

    return handler(event, context)


@middleware_before
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    try:
        payment_payload: dict = query(data=event, envelope=envelopes.EVENTBRIDGE)
        return {
            "order": Payment(**payment_payload).__dict__,
            "message": "payment created",
            "success": True,
        }
    except Exception as e:
        raise PaymentError("Unable to create payment") from e
```

```
{
    "version": "0",
    "id": "9c95e8e4-96a4-ef3f-b739-b6aa5b193afb",
    "detail-type": "PaymentCreated",
    "source": "app.payment",
    "account": "0123456789012",
    "time": "2022-08-08T20:41:53Z",
    "region": "eu-east-1",
    "detail": {
      "amount": "150.00",
      "order_id": "8f1f1710-1b30-48a5-a6bd-153fd23b866b",
      "user_id": "f80e3c51-5b8c-49d5-af7d-c7804966235f"
    }
  }
```

### Middleware with after logic

```
import time
from collections.abc import Callable

import requests
from requests import Response

from aws_lambda_powertools.event_handler import APIGatewayRestResolver
from aws_lambda_powertools.middleware_factory import lambda_handler_decorator
from aws_lambda_powertools.utilities.typing import LambdaContext

app = APIGatewayRestResolver()


@lambda_handler_decorator
def middleware_after(
    handler: Callable[[dict, LambdaContext], dict],
    event: dict,
    context: LambdaContext,
) -> dict:
    start_time = time.time()
    response = handler(event, context)
    execution_time = time.time() - start_time

    # adding custom headers in response object after lambda executing
    response["headers"]["execution_time"] = execution_time
    response["headers"]["aws_request_id"] = context.aws_request_id

    return response


@app.post("/todos")
def create_todo() -> dict:
    todo_data: dict = app.current_event.json_body  # deserialize json str to dict
    todo: Response = requests.post("https://jsonplaceholder.typicode.com/todos", data=todo_data)
    todo.raise_for_status()

    return {"todo": todo.json()}


@middleware_after
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    return app.resolve(event, context)
```

```
{
    "resource": "/todos",
    "path": "/todos",
    "httpMethod": "POST",
    "body": "{\"title\": \"foo\", \"userId\": 1, \"completed\": false}"
}
```

### Middleware with params

You can also have your own keyword arguments after the mandatory arguments.

```
import base64
from collections.abc import Callable
from dataclasses import dataclass, field
from typing import Any
from uuid import uuid4

from aws_lambda_powertools.middleware_factory import lambda_handler_decorator
from aws_lambda_powertools.utilities.jmespath_utils import (
    envelopes,
    query,
)
from aws_lambda_powertools.utilities.typing import LambdaContext


@dataclass
class Booking:
    days: int
    date_from: str
    date_to: str
    hotel_id: int
    country: str
    city: str
    guest: dict
    booking_id: str = field(default_factory=lambda: f"{uuid4()}")


class BookingError(Exception): ...


@lambda_handler_decorator
def obfuscate_sensitive_data(
    handler: Callable[[dict, LambdaContext], dict],
    event: dict,
    context: LambdaContext,
    fields: list,
) -> dict:
    # extracting payload from a EventBridge event
    detail: dict = query(data=event, envelope=envelopes.EVENTBRIDGE)
    guest_data: Any = detail.get("guest")

    # Obfuscate fields (email, vat, passport) before calling Lambda handler
    for guest_field in fields:
        if guest_data.get(guest_field):
            event["detail"]["guest"][guest_field] = obfuscate_data(str(guest_data.get(guest_field)))

    return handler(event, context)


def obfuscate_data(value: str) -> bytes:
    # base64 is not effective for obfuscation, this is an example
    return base64.b64encode(value.encode("ascii"))


@obfuscate_sensitive_data(fields=["email", "passport", "vat"])
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    try:
        booking_payload: dict = query(data=event, envelope=envelopes.EVENTBRIDGE)
        return {
            "book": Booking(**booking_payload).__dict__,
            "message": "booking created",
            "success": True,
        }
    except Exception as e:
        raise BookingError("Unable to create booking") from e
```

```
{
    "version": "0",
    "id": "9c95e8e4-96a4-ef3f-b739-b6aa5b193afb",
    "detail-type": "BookingCreated",
    "source": "app.booking",
    "account": "0123456789012",
    "time": "2022-08-08T20:41:53Z",
    "region": "eu-east-1",
    "detail": {
      "days": 5,
      "date_from": "2020-08-08",
      "date_to": "2020-08-13",
      "hotel_id": "1",
      "country": "Portugal",
      "city": "Lisbon",
      "guest": {
        "name": "Lambda",
        "email": "lambda@powertool.tools",
        "passport": "AA123456",
        "vat": "123456789"
      }
    }
}
```

### Environment variables

The following environment variable is available to configure the middleware factory at a global scope:

| Setting | Description | Environment variable | Default |
| --- | --- | --- | --- |
| **Middleware Trace** | Creates sub-segment for each custom middleware. | `POWERTOOLS_TRACE_MIDDLEWARES` | `false` |

You can also use [`POWERTOOLS_TRACE_MIDDLEWARES`](<#tracing-middleware-execution>) on a per-method basis, which will consequently override the environment variable value.

## Advanced

For advanced use cases, you can instantiate [Tracer](<https://docs.aws.amazon.com/powertools/python/latest/core/tracer/index.md>) inside your middleware, and add annotations as well as metadata for additional operational insights.

```
import time
from collections.abc import Callable

import requests
from requests import Response

from aws_lambda_powertools import Tracer
from aws_lambda_powertools.event_handler import APIGatewayRestResolver
from aws_lambda_powertools.middleware_factory import lambda_handler_decorator
from aws_lambda_powertools.utilities.typing import LambdaContext

tracer = Tracer()
app = APIGatewayRestResolver()


@lambda_handler_decorator(trace_execution=True)
def middleware_with_advanced_tracing(
    handler: Callable[[dict, LambdaContext], dict],
    event: dict,
    context: LambdaContext,
) -> dict:
    tracer.put_metadata(key="resource", value=event.get("resource"))

    start_time = time.time()
    response = handler(event, context)
    execution_time = time.time() - start_time

    tracer.put_annotation(key="TotalExecutionTime", value=str(execution_time))

    # adding custom headers in response object after lambda executing
    response["headers"]["execution_time"] = execution_time
    response["headers"]["aws_request_id"] = context.aws_request_id

    return response


@app.get("/products")
def create_product() -> dict:
    product: Response = requests.get("https://dummyjson.com/products/1")
    product.raise_for_status()

    return {"product": product.json()}


@middleware_with_advanced_tracing
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    return app.resolve(event, context)
```

```
{
    "resource": "/products",
    "path": "/products",
    "httpMethod": "GET"
 }
```

### Tracing middleware **execution**

If you are making use of [Tracer](<https://docs.aws.amazon.com/powertools/python/latest/core/tracer/index.md>), you can trace the execution of your middleware to ease operations.

This makes use of an existing Tracer instance that you may have initialized anywhere in your code.

> [!WARNING]
>
> You must [enable Active Tracing](<https://docs.aws.amazon.com/powertools/python/latest/core/tracer/#permissions>) in your Lambda function when using this feature, otherwise Lambda cannot send traces to XRay.

```
import time
from collections.abc import Callable

import requests
from requests import Response

from aws_lambda_powertools.event_handler import APIGatewayRestResolver
from aws_lambda_powertools.middleware_factory import lambda_handler_decorator
from aws_lambda_powertools.utilities.typing import LambdaContext

app = APIGatewayRestResolver()


@lambda_handler_decorator(trace_execution=True)
def middleware_with_tracing(
    handler: Callable[[dict, LambdaContext], dict],
    event: dict,
    context: LambdaContext,
) -> dict:
    start_time = time.time()
    response = handler(event, context)
    execution_time = time.time() - start_time

    # adding custom headers in response object after lambda executing
    response["headers"]["execution_time"] = execution_time
    response["headers"]["aws_request_id"] = context.aws_request_id

    return response


@app.get("/products")
def create_product() -> dict:
    product: Response = requests.get("https://dummyjson.com/products/1")
    product.raise_for_status()

    return {"product": product.json()}


@middleware_with_tracing
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    return app.resolve(event, context)
```

```
{
    "resource": "/products",
    "path": "/products",
    "httpMethod": "GET"
 }
```

When executed, your middleware name will [appear in AWS X-Ray Trace details as](<https://docs.aws.amazon.com/powertools/python/latest/core/tracer/index.md>) `## middleware_name`, in this example the middleware name is `## middleware_with_tracing`.

### Combining Powertools for AWS Lambda (Python) utilities

You can create your own middleware and combine many features of Powertools for AWS Lambda (Python) such as [trace](<https://docs.aws.amazon.com/powertools/python/latest/core/logger/index.md>), [logs](<https://docs.aws.amazon.com/powertools/python/latest/core/logger/index.md>), [feature flags](<https://docs.aws.amazon.com/powertools/python/latest/utilities/feature_flags/index.md>), [validation](<https://docs.aws.amazon.com/powertools/python/latest/utilities/validation/index.md>), [jmespath\_functions](<https://docs.aws.amazon.com/powertools/python/latest/utilities/jmespath_functions/index.md>) and others to abstract non-functional code.

In the example below, we create a Middleware with the following features:

- Logs and traces
- Validate if the payload contains a specific header
- Extract specific keys from event
- Automatically add security headers on every execution
- Validate if a specific feature flag is enabled
- Save execution history to a DynamoDB table

```
import json
from collections.abc import Callable
from urllib.parse import quote

import boto3
import combining_powertools_utilities_schema as schemas
import requests

from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.event_handler import APIGatewayRestResolver
from aws_lambda_powertools.event_handler.exceptions import InternalServerError
from aws_lambda_powertools.middleware_factory import lambda_handler_decorator
from aws_lambda_powertools.utilities.feature_flags import AppConfigStore, FeatureFlags
from aws_lambda_powertools.utilities.feature_flags.types import JSONType
from aws_lambda_powertools.utilities.jmespath_utils import query
from aws_lambda_powertools.utilities.typing import LambdaContext
from aws_lambda_powertools.utilities.validation import SchemaValidationError, validate

app = APIGatewayRestResolver()
tracer = Tracer()
logger = Logger()

table_historic = boto3.resource("dynamodb").Table("HistoricTable")

app_config = AppConfigStore(environment="dev", application="comments", name="features")
feature_flags = FeatureFlags(store=app_config)


@lambda_handler_decorator(trace_execution=True)
def middleware_custom(
    handler: Callable[[dict, LambdaContext], dict],
    event: dict,
    context: LambdaContext,
) -> dict:
    # validating the INPUT with the given schema
    # X-Customer-Id header must be informed in all requests
    try:
        validate(event=event, schema=schemas.INPUT)
    except SchemaValidationError as e:
        return {
            "statusCode": 400,
            "body": json.dumps(str(e)),
        }

    # extracting headers and requestContext from event
    headers = query(data=event, envelope="headers")
    request_context = query(data=event, envelope="requestContext")

    logger.debug(f"X-Customer-Id => {headers.get('X-Customer-Id')}")
    tracer.put_annotation(key="CustomerId", value=headers.get("X-Customer-Id"))

    response = handler(event, context)

    # automatically adding security headers to all responses
    # see: https://securityheaders.com/
    logger.info("Injecting security headers")
    response["headers"]["Referrer-Policy"] = "no-referrer"
    response["headers"]["Strict-Transport-Security"] = "max-age=15552000; includeSubDomains; preload"
    response["headers"]["X-DNS-Prefetch-Control"] = "off"
    response["headers"]["X-Content-Type-Options"] = "nosniff"
    response["headers"]["X-Permitted-Cross-Domain-Policies"] = "none"
    response["headers"]["X-Download-Options"] = "noopen"

    logger.info("Saving api call in history table")
    save_api_execution_history(str(event.get("path")), headers, request_context)

    # return lambda execution
    return response


@tracer.capture_method
def save_api_execution_history(path: str, headers: dict, request_context: dict) -> None:
    try:
        # using the feature flags utility to check if the new feature "save api call to history" is enabled by default
        # see: https://docs.aws.amazon.com/powertools/python/latest/utilities/feature_flags/#static-flags
        save_history: JSONType = feature_flags.evaluate(name="save_history", default=False)
        if save_history:
            # saving history in dynamodb table
            tracer.put_metadata(key="execution detail", value=request_context)
            table_historic.put_item(
                Item={
                    "customer_id": headers.get("X-Customer-Id"),
                    "request_id": request_context.get("requestId"),
                    "path": path,
                    "request_time": request_context.get("requestTime"),
                    "source_ip": request_context.get("identity", {}).get("sourceIp"),
                    "http_method": request_context.get("httpMethod"),
                },
            )

        return None
    except Exception:
        # you can add more logic here to handle exceptions or even save this to a DLQ
        # but not to make this example too long, we just return None since the Lambda has been successfully executed
        return None


@app.get("/comments")
@tracer.capture_method
def get_comments():
    try:
        comments: requests.Response = requests.get("https://jsonplaceholder.typicode.com/comments")
        comments.raise_for_status()

        return {"comments": comments.json()[:10]}
    except Exception as exc:
        raise InternalServerError(str(exc)) from exc


@app.get("/comments/<comment_id>")
@tracer.capture_method
def get_comments_by_id(comment_id: str):
    try:
        comment_id = quote(comment_id, safe="")
        comments: requests.Response = requests.get(f"https://jsonplaceholder.typicode.com/comments/{comment_id}")
        comments.raise_for_status()

        return {"comments": comments.json()}
    except Exception as exc:
        raise InternalServerError(str(exc)) from exc


@middleware_custom
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    return app.resolve(event, context)
```

```
INPUT = {
    "$schema": "http://json-schema.org/draft-07/schema#",
    "$id": "https://example.com/object1661012141.json",
    "title": "Root",
    "type": "object",
    "required": ["headers"],
    "properties": {
        "headers": {
            "$id": "#root/headers",
            "title": "Headers",
            "type": "object",
            "required": ["X-Customer-Id"],
            "properties": {
                "X-Customer-Id": {
                    "$id": "#root/headers/X-Customer-Id",
                    "title": "X-customer-id",
                    "type": "string",
                    "default": "",
                    "examples": ["1"],
                    "pattern": "^.*$",
                },
            },
        },
    },
}
```

```
{
    "body":"None",
    "headers":{
       "Accept":"*/*",
       "Accept-Encoding":"gzip, deflate, br",
       "Connection":"keep-alive",
       "Host":"127.0.0.1:3001",
       "Postman-Token":"a9d49365-ebe1-4bb0-8627-d5e37cdce86d",
       "User-Agent":"PostmanRuntime/7.29.0",
       "X-Customer-Id":"1",
       "X-Forwarded-Port":"3001",
       "X-Forwarded-Proto":"http"
    },
    "httpMethod":"GET",
    "isBase64Encoded":false,
    "multiValueHeaders":{
       "Accept":[
          "*/*"
       ],
       "Accept-Encoding":[
          "gzip, deflate, br"
       ],
       "Connection":[
          "keep-alive"
       ],
       "Host":[
          "127.0.0.1:3001"
       ],
       "Postman-Token":[
          "a9d49365-ebe1-4bb0-8627-d5e37cdce86d"
       ],
       "User-Agent":[
          "PostmanRuntime/7.29.0"
       ],
       "X-Customer-Id":[
          "1"
       ],
       "X-Forwarded-Port":[
          "3001"
       ],
       "X-Forwarded-Proto":[
          "http"
       ]
    },
    "multiValueQueryStringParameters":"None",
    "path":"/comments",
    "pathParameters":"None",
    "queryStringParameters":"None",
    "requestContext":{
       "accountId":"123456789012",
       "apiId":"1234567890",
       "domainName":"127.0.0.1:3001",
       "extendedRequestId":"None",
       "httpMethod":"GET",
       "identity":{
          "accountId":"None",
          "apiKey":"None",
          "caller":"None",
          "cognitoAuthenticationProvider":"None",
          "cognitoAuthenticationType":"None",
          "cognitoIdentityPoolId":"None",
          "sourceIp":"127.0.0.1",
          "user":"None",
          "userAgent":"Custom User Agent String",
          "userArn":"None"
       },
       "path":"/comments",
       "protocol":"HTTP/1.1",
       "requestId":"56d1a102-6d9d-4f13-b4f7-26751c10a131",
       "requestTime":"20/Aug/2022:18:18:58 +0000",
       "requestTimeEpoch":1661019538,
       "resourceId":"123456",
       "resourcePath":"/comments",
       "stage":"Prod"
    },
    "resource":"/comments",
    "stageVariables":"None",
    "version":"1.0"
 }
```

```
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: Middleware-powertools-utilities example

Globals:
  Function:
    Timeout: 5
    Runtime: python3.12
    Tracing: Active
    Architectures:
      - x86_64
    Environment:
      Variables:
        POWERTOOLS_LOG_LEVEL: DEBUG
        POWERTOOLS_LOGGER_SAMPLE_RATE: 0.1
        POWERTOOLS_LOGGER_LOG_EVENT: true
        POWERTOOLS_SERVICE_NAME: middleware

Resources:
  MiddlewareFunction:
    Type: AWS::Serverless::Function # More info about Function Resource: https://github.com/awslabs/serverless-application-model/blob/master/versions/2016-10-31.md#awsserverlessfunction
    Properties:
      CodeUri: middleware/
      Handler: app.lambda_handler
      Description: Middleware function
      Policies:
      - AWSLambdaBasicExecutionRole # Managed Policy
      - Version: '2012-10-17' # Policy Document
        Statement:
          - Effect: Allow
            Action:
              - dynamodb:PutItem
            Resource: !GetAtt HistoryTable.Arn
          - Effect: Allow
            Action: # https://docs.aws.amazon.com/appconfig/latest/userguide/getting-started-with-appconfig-permissions.html
              - ssm:GetDocument
              - ssm:ListDocuments
              - appconfig:GetLatestConfiguration
              - appconfig:StartConfigurationSession
              - appconfig:ListApplications
              - appconfig:GetApplication
              - appconfig:ListEnvironments
              - appconfig:GetEnvironment
              - appconfig:ListConfigurationProfiles
              - appconfig:GetConfigurationProfile
              - appconfig:ListDeploymentStrategies
              - appconfig:GetDeploymentStrategy
              - appconfig:GetConfiguration
              - appconfig:ListDeployments
              - appconfig:GetDeployment
            Resource: "*"
      Events:
        GetComments:
          Type: Api
          Properties:
            Path: /comments
            Method: GET
        GetCommentsById:
          Type: Api
          Properties:
            Path: /comments/{comment_id}
            Method: GET

  # DynamoDB table to store historical data
  HistoryTable:
    Type: AWS::DynamoDB::Table
    Properties:
      TableName: "HistoryTable"
      AttributeDefinitions:
        - AttributeName: customer_id
          AttributeType: S
        - AttributeName: request_id
          AttributeType: S
      KeySchema:
        - AttributeName: customer_id
          KeyType: HASH
        - AttributeName: request_id
          KeyType: "RANGE"
      BillingMode: PAY_PER_REQUEST

  # Feature flags using AppConfig
  FeatureCommentApp:
    Type: AWS::AppConfig::Application
    Properties:
      Description: "Comments Application for feature toggles"
      Name: comments

  FeatureCommentDevEnv:
    Type: AWS::AppConfig::Environment
    Properties:
      ApplicationId: !Ref FeatureCommentApp
      Description: "Development Environment for the App Config Comments"
      Name: dev

  FeatureCommentConfigProfile:
    Type: AWS::AppConfig::ConfigurationProfile
    Properties:
      ApplicationId: !Ref FeatureCommentApp
      Name: features
      LocationUri: "hosted"

  HostedConfigVersion:
    Type: AWS::AppConfig::HostedConfigurationVersion
    Properties:
      ApplicationId: !Ref FeatureCommentApp
      ConfigurationProfileId: !Ref FeatureCommentConfigProfile
      Description: 'A sample hosted configuration version'
      Content: |
        {
              "save_history": {
                "default": true
              }
        }
      ContentType: 'application/json'

  # this is just an example
  # change this values according your deployment strategy
  BasicDeploymentStrategy:
    Type: AWS::AppConfig::DeploymentStrategy
    Properties:
      Name: "Deployment"
      Description: "Deployment strategy for comments app."
      DeploymentDurationInMinutes: 1
      FinalBakeTimeInMinutes: 1
      GrowthFactor: 100
      GrowthType: LINEAR
      ReplicateTo: NONE

  ConfigDeployment:
    Type: AWS::AppConfig::Deployment
    Properties:
      ApplicationId: !Ref FeatureCommentApp
      ConfigurationProfileId: !Ref FeatureCommentConfigProfile
      ConfigurationVersion: !Ref HostedConfigVersion
      DeploymentStrategyId: !Ref BasicDeploymentStrategy
      EnvironmentId: !Ref FeatureCommentDevEnv
```

## Tips

- Use `trace_execution` to quickly understand the performance impact of your middlewares, and reduce or merge tasks when necessary
- When nesting multiple middlewares, always return the handler with event and context, or response
- Keep in mind [Python decorators execution order](<https://realpython.com/primer-on-python-decorators/#nesting-decorators>). Lambda handler is actually called once (top-down)
- Async middlewares are not supported