REST

CLASS	DESCRIPTION
ALBResolver	Amazon Application Load Balancer (ALB) resolver
APIGatewayHttpResolver	Amazon API Gateway HTTP API v2 payload resolver
APIGatewayRestResolver	Amazon API Gateway REST and HTTP API v1 payload resolver
ApiGatewayResolver	API Gateway, VPC Laticce, Bedrock and ALB proxy resolver
BaseRouter	Base class for Routing
CORSConfig	CORS Config
MiddlewareFrame	Creates a Middle Stack Wrapper instance to be used as a "Frame" in the overall stack of
ProxyEventType	An enumerations of the supported proxy event types.
Response	Response data class that provides greater control over what is returned from the proxy event
ResponseBuilder	Internally used Response builder
Route	Internally used Route Configuration
Router	Router helper class to allow splitting ApiGatewayResolver into multiple files

ALBResolver

ALBResolver(
    cors: CORSConfig | None = None,
    debug: bool | None = None,
    serializer: Callable[[dict], str] | None = None,
    strip_prefixes: list[str | Pattern] | None = None,
    enable_validation: bool = False,
)

Bases: ApiGatewayResolver

Amazon Application Load Balancer (ALB) resolver

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def __init__(
    self,
    cors: CORSConfig | None = None,
    debug: bool | None = None,
    serializer: Callable[[dict], str] | None = None,
    strip_prefixes: list[str | Pattern] | None = None,
    enable_validation: bool = False,
):
    """Amazon Application Load Balancer (ALB) resolver"""
    super().__init__(ProxyEventType.ALBEvent, cors, debug, serializer, strip_prefixes, enable_validation)

APIGatewayHttpResolver

APIGatewayHttpResolver(
    cors: CORSConfig | None = None,
    debug: bool | None = None,
    serializer: Callable[[dict], str] | None = None,
    strip_prefixes: list[str | Pattern] | None = None,
    enable_validation: bool = False,
)

Bases: ApiGatewayResolver

Amazon API Gateway HTTP API v2 payload resolver

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def __init__(
    self,
    cors: CORSConfig | None = None,
    debug: bool | None = None,
    serializer: Callable[[dict], str] | None = None,
    strip_prefixes: list[str | Pattern] | None = None,
    enable_validation: bool = False,
):
    """Amazon API Gateway HTTP API v2 payload resolver"""
    super().__init__(
        ProxyEventType.APIGatewayProxyEventV2,
        cors,
        debug,
        serializer,
        strip_prefixes,
        enable_validation,
    )

APIGatewayRestResolver

APIGatewayRestResolver(
    cors: CORSConfig | None = None,
    debug: bool | None = None,
    serializer: Callable[[dict], str] | None = None,
    strip_prefixes: list[str | Pattern] | None = None,
    enable_validation: bool = False,
)

Bases: ApiGatewayResolver

Amazon API Gateway REST and HTTP API v1 payload resolver

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def __init__(
    self,
    cors: CORSConfig | None = None,
    debug: bool | None = None,
    serializer: Callable[[dict], str] | None = None,
    strip_prefixes: list[str | Pattern] | None = None,
    enable_validation: bool = False,
):
    """Amazon API Gateway REST and HTTP API v1 payload resolver"""
    super().__init__(
        ProxyEventType.APIGatewayProxyEvent,
        cors,
        debug,
        serializer,
        strip_prefixes,
        enable_validation,
    )

ApiGatewayResolver

ApiGatewayResolver(
    proxy_type: Enum = ProxyEventType.APIGatewayProxyEvent,
    cors: CORSConfig | None = None,
    debug: bool | None = None,
    serializer: Callable[[dict], str] | None = None,
    strip_prefixes: list[str | Pattern] | None = None,
    enable_validation: bool = False,
)

Bases: BaseRouter

API Gateway, VPC Laticce, Bedrock and ALB proxy resolver

Examples:

Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator

from aws_lambda_powertools import Tracer
from aws_lambda_powertools.event_handler import APIGatewayRestResolver

tracer = Tracer()
app = APIGatewayRestResolver()

@app.get("/get-call")
def simple_get():
    return {"message": "Foo"}

@app.post("/post-call")
def simple_post():
    post_data: dict = app.current_event.json_body
    return {"message": post_data["value"]}

@tracer.capture_lambda_handler
def lambda_handler(event, context):
    return app.resolve(event, context)

METHOD	DESCRIPTION
enable_swagger	Returns the OpenAPI schema as a JSON serializable dict
get_openapi_json_schema	Returns the OpenAPI schema as a JSON serializable dict
get_openapi_schema	Returns the OpenAPI schema as a pydantic model.
include_router	Adds all routes and context defined in a router
resolve	Resolves the response based on the provide event and decorator routes
route	Route decorator includes parameter method

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def __init__(
    self,
    proxy_type: Enum = ProxyEventType.APIGatewayProxyEvent,
    cors: CORSConfig | None = None,
    debug: bool | None = None,
    serializer: Callable[[dict], str] | None = None,
    strip_prefixes: list[str | Pattern] | None = None,
    enable_validation: bool = False,
):
    """
    Parameters
    ----------
    proxy_type: ProxyEventType
        Proxy request type, defaults to API Gateway V1
    cors: CORSConfig
        Optionally configure and enabled CORS. Not each route will need to have to cors=True
    debug: bool | None
        Enables debug mode, by default False. Can be also be enabled by "POWERTOOLS_DEV"
        environment variable
    serializer: Callable, optional
        function to serialize `obj` to a JSON formatted `str`, by default json.dumps
    strip_prefixes: list[str | Pattern], optional
        optional list of prefixes to be removed from the request path before doing the routing.
        This is often used with api gateways with multiple custom mappings.
        Each prefix can be a static string or a compiled regex pattern
    enable_validation: bool | None
        Enables validation of the request body against the route schema, by default False.
    """
    self._proxy_type = proxy_type
    self._dynamic_routes: list[Route] = []
    self._static_routes: list[Route] = []
    self._route_keys: list[str] = []
    self._exception_handlers: dict[type, Callable] = {}
    self._cors = cors
    self._cors_enabled: bool = cors is not None
    self._cors_methods: set[str] = {"OPTIONS"}
    self._debug = self._has_debug(debug)
    self._enable_validation = enable_validation
    self._strip_prefixes = strip_prefixes
    self.context: dict = {}  # early init as customers might add context before event resolution
    self.processed_stack_frames = []
    self._response_builder_class = ResponseBuilder[BaseProxyEvent]

    # Allow for a custom serializer or a concise json serialization
    self._serializer = serializer or partial(json.dumps, separators=(",", ":"), cls=Encoder)

    if self._enable_validation:
        from aws_lambda_powertools.event_handler.middlewares.openapi_validation import OpenAPIValidationMiddleware

        # Note the serializer argument: only use custom serializer if provided by the caller
        # Otherwise, fully rely on the internal Pydantic based mechanism to serialize responses for validation.
        self.use([OpenAPIValidationMiddleware(validation_serializer=serializer)])

enable_swagger

enable_swagger(
    *,
    path: str = "/swagger",
    title: str = "Powertools for AWS Lambda (Python) API",
    version: str = DEFAULT_API_VERSION,
    openapi_version: str = DEFAULT_OPENAPI_VERSION,
    summary: str | None = None,
    description: str | None = None,
    tags: list[Tag | str] | None = None,
    servers: list[Server] | None = None,
    terms_of_service: str | None = None,
    contact: Contact | None = None,
    license_info: License | None = None,
    swagger_base_url: str | None = None,
    middlewares: (
        list[Callable[..., Response]] | None
    ) = None,
    compress: bool = False,
    security_schemes: (
        dict[str, SecurityScheme] | None
    ) = None,
    security: list[dict[str, list[str]]] | None = None,
    oauth2_config: OAuth2Config | None = None,
    persist_authorization: bool = False,
    openapi_extensions: dict[str, Any] | None = None
)

Returns the OpenAPI schema as a JSON serializable dict

PARAMETER	DESCRIPTION
path	The path to the swagger UI. TYPE: str DEFAULT: '/swagger'
title	The title of the application. TYPE: str DEFAULT: 'Powertools for AWS Lambda (Python) API'
version	The version of the OpenAPI document (which is distinct from the OpenAPI Specification version or the API TYPE: str DEFAULT: DEFAULT_API_VERSION
openapi_version	The version of the OpenAPI Specification (which the document uses). TYPE: str DEFAULT: DEFAULT_OPENAPI_VERSION
summary	A short summary of what the application does. TYPE: str | None DEFAULT: None
description	A verbose explanation of the application behavior. TYPE: str | None DEFAULT: None
tags	A list of tags used by the specification with additional metadata. TYPE: list[Tag | str] | None DEFAULT: None
servers	An array of Server Objects, which provide connectivity information to a target server. TYPE: list[Server] | None DEFAULT: None
terms_of_service	A URL to the Terms of Service for the API. MUST be in the format of a URL. TYPE: str | None DEFAULT: None
contact	The contact information for the exposed API. TYPE: Contact | None DEFAULT: None
license_info	The license information for the exposed API. TYPE: License | None DEFAULT: None
swagger_base_url	The base url for the swagger UI. If not provided, we will serve a recent version of the Swagger UI. TYPE: str | None DEFAULT: None
middlewares	List of middlewares to be used for the swagger route. TYPE: list[Callable[..., Response]] | None DEFAULT: None
compress	Whether or not to enable gzip compression swagger route. TYPE: bool DEFAULT: False
security_schemes	A declaration of the security schemes available to be used in the specification. TYPE: dict[str, SecurityScheme] | None DEFAULT: None
security	A declaration of which security mechanisms are applied globally across the API. TYPE: list[dict[str, list[str]]] | None DEFAULT: None
oauth2_config	The OAuth2 configuration for the Swagger UI. TYPE: OAuth2Config | None DEFAULT: None
persist_authorization	Whether to persist authorization data on browser close/refresh. TYPE: bool DEFAULT: False
openapi_extensions	Additional OpenAPI extensions as a dictionary. TYPE: dict[str, Any] | None DEFAULT: None

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def enable_swagger(
    self,
    *,
    path: str = "/swagger",
    title: str = "Powertools for AWS Lambda (Python) API",
    version: str = DEFAULT_API_VERSION,
    openapi_version: str = DEFAULT_OPENAPI_VERSION,
    summary: str | None = None,
    description: str | None = None,
    tags: list[Tag | str] | None = None,
    servers: list[Server] | None = None,
    terms_of_service: str | None = None,
    contact: Contact | None = None,
    license_info: License | None = None,
    swagger_base_url: str | None = None,
    middlewares: list[Callable[..., Response]] | None = None,
    compress: bool = False,
    security_schemes: dict[str, SecurityScheme] | None = None,
    security: list[dict[str, list[str]]] | None = None,
    oauth2_config: OAuth2Config | None = None,
    persist_authorization: bool = False,
    openapi_extensions: dict[str, Any] | None = None,
):
    """
    Returns the OpenAPI schema as a JSON serializable dict

    Parameters
    ----------
    path: str, default = "/swagger"
        The path to the swagger UI.
    title: str
        The title of the application.
    version: str
        The version of the OpenAPI document (which is distinct from the OpenAPI Specification version or the API
    openapi_version: str, default = "3.0.0"
        The version of the OpenAPI Specification (which the document uses).
    summary: str, optional
        A short summary of what the application does.
    description: str, optional
        A verbose explanation of the application behavior.
    tags: list[Tag, str], optional
        A list of tags used by the specification with additional metadata.
    servers: list[Server], optional
        An array of Server Objects, which provide connectivity information to a target server.
    terms_of_service: str, optional
        A URL to the Terms of Service for the API. MUST be in the format of a URL.
    contact: Contact, optional
        The contact information for the exposed API.
    license_info: License, optional
        The license information for the exposed API.
    swagger_base_url: str, optional
        The base url for the swagger UI. If not provided, we will serve a recent version of the Swagger UI.
    middlewares: list[Callable[..., Response]], optional
        List of middlewares to be used for the swagger route.
    compress: bool, default = False
        Whether or not to enable gzip compression swagger route.
    security_schemes: dict[str, "SecurityScheme"], optional
        A declaration of the security schemes available to be used in the specification.
    security: list[dict[str, list[str]]], optional
        A declaration of which security mechanisms are applied globally across the API.
    oauth2_config: OAuth2Config, optional
        The OAuth2 configuration for the Swagger UI.
    persist_authorization: bool, optional
        Whether to persist authorization data on browser close/refresh.
    openapi_extensions: dict[str, Any], optional
        Additional OpenAPI extensions as a dictionary.
    """
    from aws_lambda_powertools.event_handler.openapi.compat import model_json
    from aws_lambda_powertools.event_handler.openapi.models import Server
    from aws_lambda_powertools.event_handler.openapi.swagger_ui import (
        generate_oauth2_redirect_html,
        generate_swagger_html,
    )

    @self.get(path, middlewares=middlewares, include_in_schema=False, compress=compress)
    def swagger_handler():
        query_params = self.current_event.query_string_parameters or {}

        # Check for query parameters; if "format" is specified as "oauth2-redirect",
        # send the oauth2-redirect HTML stanza so OAuth2 can be used
        # Source: https://github.com/swagger-api/swagger-ui/blob/master/dist/oauth2-redirect.html
        if query_params.get("format") == "oauth2-redirect":
            return Response(
                status_code=200,
                content_type="text/html",
                body=generate_oauth2_redirect_html(),
            )

        base_path = self._get_base_path()

        if swagger_base_url:
            swagger_js = f"{swagger_base_url}/swagger-ui-bundle.min.js"
            swagger_css = f"{swagger_base_url}/swagger-ui.min.css"
        else:
            # We now inject CSS and JS into the SwaggerUI file
            swagger_js = Path.open(
                Path(__file__).parent / "openapi" / "swagger_ui" / "swagger-ui-bundle.min.js",
            ).read()
            swagger_css = Path.open(Path(__file__).parent / "openapi" / "swagger_ui" / "swagger-ui.min.css").read()

        openapi_servers = servers or [Server(url=(base_path or "/"))]

        spec = self.get_openapi_schema(
            title=title,
            version=version,
            openapi_version=openapi_version,
            summary=summary,
            description=description,
            tags=tags,
            servers=openapi_servers,
            terms_of_service=terms_of_service,
            contact=contact,
            license_info=license_info,
            security_schemes=security_schemes,
            security=security,
            openapi_extensions=openapi_extensions,
        )

        # The .replace('</', '<\\/') part is necessary to prevent a potential issue where the JSON string contains
        # </script> or similar tags. Escaping the forward slash in </ as <\/ ensures that the JSON does not
        # inadvertently close the script tag, and the JSON remains a valid string within the JavaScript code.
        escaped_spec = model_json(
            spec,
            by_alias=True,
            exclude_none=True,
            indent=2,
        ).replace("</", "<\\/")

        # Check for query parameters; if "format" is specified as "json",
        # respond with the JSON used in the OpenAPI spec
        # Example: https://www.example.com/swagger?format=json
        if query_params.get("format") == "json":
            return Response(
                status_code=200,
                content_type="application/json",
                body=escaped_spec,
            )

        body = generate_swagger_html(
            escaped_spec,
            swagger_js,
            swagger_css,
            swagger_base_url,
            oauth2_config,
            persist_authorization,
        )

        return Response(
            status_code=200,
            content_type="text/html",
            body=body,
        )

get_openapi_json_schema

get_openapi_json_schema(
    *,
    title: str = "Powertools API",
    version: str = DEFAULT_API_VERSION,
    openapi_version: str = DEFAULT_OPENAPI_VERSION,
    summary: str | None = None,
    description: str | None = None,
    tags: list[Tag | str] | None = None,
    servers: list[Server] | None = None,
    terms_of_service: str | None = None,
    contact: Contact | None = None,
    license_info: License | None = None,
    security_schemes: (
        dict[str, SecurityScheme] | None
    ) = None,
    security: list[dict[str, list[str]]] | None = None,
    openapi_extensions: dict[str, Any] | None = None
) -> str

Returns the OpenAPI schema as a JSON serializable dict

PARAMETER	DESCRIPTION
title	The title of the application. TYPE: str DEFAULT: 'Powertools API'
version	The version of the OpenAPI document (which is distinct from the OpenAPI Specification version or the API TYPE: str DEFAULT: DEFAULT_API_VERSION
openapi_version	The version of the OpenAPI Specification (which the document uses). TYPE: str DEFAULT: DEFAULT_OPENAPI_VERSION
summary	A short summary of what the application does. TYPE: str | None DEFAULT: None
description	A verbose explanation of the application behavior. TYPE: str | None DEFAULT: None
tags	A list of tags used by the specification with additional metadata. TYPE: list[Tag | str] | None DEFAULT: None
servers	An array of Server Objects, which provide connectivity information to a target server. TYPE: list[Server] | None DEFAULT: None
terms_of_service	A URL to the Terms of Service for the API. MUST be in the format of a URL. TYPE: str | None DEFAULT: None
contact	The contact information for the exposed API. TYPE: Contact | None DEFAULT: None
license_info	The license information for the exposed API. TYPE: License | None DEFAULT: None
security_schemes	A declaration of the security schemes available to be used in the specification. TYPE: dict[str, SecurityScheme] | None DEFAULT: None
security	A declaration of which security mechanisms are applied globally across the API. TYPE: list[dict[str, list[str]]] | None DEFAULT: None
openapi_extensions	Additional OpenAPI extensions as a dictionary. TYPE: dict[str, Any] | None DEFAULT: None

RETURNS	DESCRIPTION
str	The OpenAPI schema as a JSON serializable dict.

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def get_openapi_json_schema(
    self,
    *,
    title: str = "Powertools API",
    version: str = DEFAULT_API_VERSION,
    openapi_version: str = DEFAULT_OPENAPI_VERSION,
    summary: str | None = None,
    description: str | None = None,
    tags: list[Tag | str] | None = None,
    servers: list[Server] | None = None,
    terms_of_service: str | None = None,
    contact: Contact | None = None,
    license_info: License | None = None,
    security_schemes: dict[str, SecurityScheme] | None = None,
    security: list[dict[str, list[str]]] | None = None,
    openapi_extensions: dict[str, Any] | None = None,
) -> str:
    """
    Returns the OpenAPI schema as a JSON serializable dict

    Parameters
    ----------
    title: str
        The title of the application.
    version: str
        The version of the OpenAPI document (which is distinct from the OpenAPI Specification version or the API
    openapi_version: str, default = "3.0.0"
        The version of the OpenAPI Specification (which the document uses).
    summary: str, optional
        A short summary of what the application does.
    description: str, optional
        A verbose explanation of the application behavior.
    tags: list[Tag, str], optional
        A list of tags used by the specification with additional metadata.
    servers: list[Server], optional
        An array of Server Objects, which provide connectivity information to a target server.
    terms_of_service: str, optional
        A URL to the Terms of Service for the API. MUST be in the format of a URL.
    contact: Contact, optional
        The contact information for the exposed API.
    license_info: License, optional
        The license information for the exposed API.
    security_schemes: dict[str, SecurityScheme]], optional
        A declaration of the security schemes available to be used in the specification.
    security: list[dict[str, list[str]]], optional
        A declaration of which security mechanisms are applied globally across the API.
    openapi_extensions: Dict[str, Any], optional
        Additional OpenAPI extensions as a dictionary.

    Returns
    -------
    str
        The OpenAPI schema as a JSON serializable dict.
    """
    from aws_lambda_powertools.event_handler.openapi.compat import model_json

    return model_json(
        self.get_openapi_schema(
            title=title,
            version=version,
            openapi_version=openapi_version,
            summary=summary,
            description=description,
            tags=tags,
            servers=servers,
            terms_of_service=terms_of_service,
            contact=contact,
            license_info=license_info,
            security_schemes=security_schemes,
            security=security,
            openapi_extensions=openapi_extensions,
        ),
        by_alias=True,
        exclude_none=True,
        indent=2,
    )

get_openapi_schema

get_openapi_schema(
    *,
    title: str = "Powertools API",
    version: str = DEFAULT_API_VERSION,
    openapi_version: str = DEFAULT_OPENAPI_VERSION,
    summary: str | None = None,
    description: str | None = None,
    tags: list[Tag | str] | None = None,
    servers: list[Server] | None = None,
    terms_of_service: str | None = None,
    contact: Contact | None = None,
    license_info: License | None = None,
    security_schemes: (
        dict[str, SecurityScheme] | None
    ) = None,
    security: list[dict[str, list[str]]] | None = None,
    openapi_extensions: dict[str, Any] | None = None
) -> OpenAPI

Returns the OpenAPI schema as a pydantic model.

PARAMETER	DESCRIPTION
title	The title of the application. TYPE: str DEFAULT: 'Powertools API'
version	The version of the OpenAPI document (which is distinct from the OpenAPI Specification version or the API TYPE: str DEFAULT: DEFAULT_API_VERSION
openapi_version	The version of the OpenAPI Specification (which the document uses). TYPE: str DEFAULT: DEFAULT_OPENAPI_VERSION
summary	A short summary of what the application does. TYPE: str | None DEFAULT: None
description	A verbose explanation of the application behavior. TYPE: str | None DEFAULT: None
tags	A list of tags used by the specification with additional metadata. TYPE: list[Tag | str] | None DEFAULT: None
servers	An array of Server Objects, which provide connectivity information to a target server. TYPE: list[Server] | None DEFAULT: None
terms_of_service	A URL to the Terms of Service for the API. MUST be in the format of a URL. TYPE: str | None DEFAULT: None
contact	The contact information for the exposed API. TYPE: Contact | None DEFAULT: None
license_info	The license information for the exposed API. TYPE: License | None DEFAULT: None
security_schemes	A declaration of the security schemes available to be used in the specification. TYPE: dict[str, SecurityScheme] | None DEFAULT: None
security	A declaration of which security mechanisms are applied globally across the API. TYPE: list[dict[str, list[str]]] | None DEFAULT: None
openapi_extensions	Additional OpenAPI extensions as a dictionary. TYPE: dict[str, Any] | None DEFAULT: None

RETURNS	DESCRIPTION
OpenAPI	The OpenAPI schema as a pydantic model. TYPE: pydantic model

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def get_openapi_schema(
    self,
    *,
    title: str = "Powertools API",
    version: str = DEFAULT_API_VERSION,
    openapi_version: str = DEFAULT_OPENAPI_VERSION,
    summary: str | None = None,
    description: str | None = None,
    tags: list[Tag | str] | None = None,
    servers: list[Server] | None = None,
    terms_of_service: str | None = None,
    contact: Contact | None = None,
    license_info: License | None = None,
    security_schemes: dict[str, SecurityScheme] | None = None,
    security: list[dict[str, list[str]]] | None = None,
    openapi_extensions: dict[str, Any] | None = None,
) -> OpenAPI:
    """
    Returns the OpenAPI schema as a pydantic model.

    Parameters
    ----------
    title: str
        The title of the application.
    version: str
        The version of the OpenAPI document (which is distinct from the OpenAPI Specification version or the API
    openapi_version: str, default = "3.0.0"
        The version of the OpenAPI Specification (which the document uses).
    summary: str, optional
        A short summary of what the application does.
    description: str, optional
        A verbose explanation of the application behavior.
    tags: list[Tag | str], optional
        A list of tags used by the specification with additional metadata.
    servers: list[Server], optional
        An array of Server Objects, which provide connectivity information to a target server.
    terms_of_service: str, optional
        A URL to the Terms of Service for the API. MUST be in the format of a URL.
    contact: Contact, optional
        The contact information for the exposed API.
    license_info: License, optional
        The license information for the exposed API.
    security_schemes: dict[str, SecurityScheme]], optional
        A declaration of the security schemes available to be used in the specification.
    security: list[dict[str, list[str]]], optional
        A declaration of which security mechanisms are applied globally across the API.
    openapi_extensions: Dict[str, Any], optional
        Additional OpenAPI extensions as a dictionary.

    Returns
    -------
    OpenAPI: pydantic model
        The OpenAPI schema as a pydantic model.
    """

    from aws_lambda_powertools.event_handler.openapi.compat import (
        GenerateJsonSchema,
        get_compat_model_name_map,
        get_definitions,
    )
    from aws_lambda_powertools.event_handler.openapi.models import OpenAPI, PathItem, Tag
    from aws_lambda_powertools.event_handler.openapi.types import (
        COMPONENT_REF_TEMPLATE,
    )

    openapi_version = self._determine_openapi_version(openapi_version)

    # Start with the bare minimum required for a valid OpenAPI schema
    info: dict[str, Any] = {"title": title, "version": version}

    optional_fields = {
        "summary": summary,
        "description": description,
        "termsOfService": terms_of_service,
        "contact": contact,
        "license": license_info,
    }

    info.update({field: value for field, value in optional_fields.items() if value})

    if not isinstance(openapi_extensions, dict):
        openapi_extensions = {}

    output: dict[str, Any] = {
        "openapi": openapi_version,
        "info": info,
        "servers": self._get_openapi_servers(servers),
        "security": self._get_openapi_security(security, security_schemes),
        **openapi_extensions,
    }

    components: dict[str, dict[str, Any]] = {}
    paths: dict[str, dict[str, Any]] = {}
    operation_ids: set[str] = set()

    all_routes = self._dynamic_routes + self._static_routes
    all_fields = self._get_fields_from_routes(all_routes)
    model_name_map = get_compat_model_name_map(all_fields)

    # Collect all models and definitions
    schema_generator = GenerateJsonSchema(ref_template=COMPONENT_REF_TEMPLATE)
    field_mapping, definitions = get_definitions(
        fields=all_fields,
        schema_generator=schema_generator,
        model_name_map=model_name_map,
    )

    # Add routes to the OpenAPI schema
    for route in all_routes:
        if route.security and not _validate_openapi_security_parameters(
            security=route.security,
            security_schemes=security_schemes,
        ):
            raise SchemaValidationError(
                "Security configuration was not found in security_schemas or security_schema was not defined. "
                "See: https://docs.powertools.aws.dev/lambda/python/latest/core/event_handler/api_gateway/#security-schemes",
            )

        if not route.include_in_schema:
            continue

        result = route._get_openapi_path(
            dependant=route.dependant,
            operation_ids=operation_ids,
            model_name_map=model_name_map,
            field_mapping=field_mapping,
        )
        if result:
            path, path_definitions = result
            if path:
                paths.setdefault(route.openapi_path, {}).update(path)
            if path_definitions:
                definitions.update(path_definitions)

    if definitions:
        components["schemas"] = {k: definitions[k] for k in sorted(definitions)}
    if security_schemes:
        components["securitySchemes"] = security_schemes
    if components:
        output["components"] = components
    if tags:
        output["tags"] = [Tag(name=tag) if isinstance(tag, str) else tag for tag in tags]

    output["paths"] = {k: PathItem(**v) for k, v in paths.items()}

    return OpenAPI(**output)

include_router

include_router(
    router: Router, prefix: str | None = None
) -> None

Adds all routes and context defined in a router

PARAMETER	DESCRIPTION
router	The Router containing a list of routes to be registered after the existing routes TYPE: Router
prefix	An optional prefix to be added to the originally defined rule TYPE: str DEFAULT: None

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def include_router(self, router: Router, prefix: str | None = None) -> None:
    """Adds all routes and context defined in a router

    Parameters
    ----------
    router : Router
        The Router containing a list of routes to be registered after the existing routes
    prefix : str, optional
        An optional prefix to be added to the originally defined rule
    """

    # Add reference to parent ApiGatewayResolver to support use cases where people subclass it to add custom logic
    router.api_resolver = self

    logger.debug("Merging App context with Router context")
    self.context.update(**router.context)

    logger.debug("Appending Router middlewares into App middlewares.")
    self._router_middlewares = self._router_middlewares + router._router_middlewares

    logger.debug("Appending Router exception_handler into App exception_handler.")
    self._exception_handlers.update(router._exception_handlers)

    # use pointer to allow context clearance after event is processed e.g., resolve(evt, ctx)
    router.context = self.context

    # Iterate through the routes defined in the router to configure and apply middlewares for each route
    for route, func in router._routes.items():
        new_route = route

        if prefix:
            rule = route[0]
            rule = prefix if rule == "/" else f"{prefix}{rule}"
            new_route = (rule, *route[1:])

        # Middlewares are stored by route separately - must grab them to include
        # Middleware store the route without prefix, so we must not include prefix when grabbing
        middlewares = router._routes_with_middleware.get(route)

        # Need to use "type: ignore" here since mypy does not like a named parameter after
        # tuple expansion since may cause duplicate named parameters in the function signature.
        # In this case this is not possible since the tuple expansion is from a hashable source
        # and the `middlewares` list is a non-hashable structure so will never be included.
        # Still need to ignore for mypy checks or will cause failures (false-positive)
        self.route(*new_route, middlewares=middlewares)(func)  # type: ignore

resolve

resolve(
    event: dict[str, Any], context: LambdaContext
) -> dict[str, Any]

Resolves the response based on the provide event and decorator routes

Internals

Request processing chain is triggered by a Route object being called (_call_route -> __call__):

1. When a route is matched 1.1. Exception handlers (if any exception bubbled up and caught) 1.2. Global middlewares (before, and after on the way back) 1.3. Path level middleware (before, and after on the way back) 1.4. Middleware adapter to ensure Response is homogenous (_registered_api_adapter) 1.5. Run actual route
2. When a route is NOT matched 2.1. Exception handlers (if any exception bubbled up and caught) 2.2. Global middlewares (before, and after on the way back) 2.3. Path level middleware (before, and after on the way back) 2.4. Middleware adapter to ensure Response is homogenous (_registered_api_adapter) 2.5. Run 404 route handler
3. When a route is a pre-flight CORS (often not matched) 3.1. Exception handlers (if any exception bubbled up and caught) 3.2. Global middlewares (before, and after on the way back) 3.3. Path level middleware (before, and after on the way back) 3.4. Middleware adapter to ensure Response is homogenous (_registered_api_adapter) 3.5. Return 204 with appropriate CORS headers
4. When a route is matched with Data Validation enabled 4.1. Exception handlers (if any exception bubbled up and caught) 4.2. Data Validation middleware (before, and after on the way back) 4.3. Global middlewares (before, and after on the way back) 4.4. Path level middleware (before, and after on the way back) 4.5. Middleware adapter to ensure Response is homogenous (_registered_api_adapter) 4.6. Run actual route

PARAMETER	DESCRIPTION
event	Event TYPE: dict[str, Any]
context	Lambda context TYPE: LambdaContext

RETURNS	DESCRIPTION
dict	Returns the dict response

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def resolve(self, event: dict[str, Any], context: LambdaContext) -> dict[str, Any]:
    """Resolves the response based on the provide event and decorator routes

    ## Internals

    Request processing chain is triggered by a Route object being called _(`_call_route` -> `__call__`)_:

    1. **When a route is matched**
      1.1. Exception handlers _(if any exception bubbled up and caught)_
      1.2. Global middlewares _(before, and after on the way back)_
      1.3. Path level middleware _(before, and after on the way back)_
      1.4. Middleware adapter to ensure Response is homogenous (_registered_api_adapter)
      1.5. Run actual route
    2. **When a route is NOT matched**
      2.1. Exception handlers _(if any exception bubbled up and caught)_
      2.2. Global middlewares _(before, and after on the way back)_
      2.3. Path level middleware _(before, and after on the way back)_
      2.4. Middleware adapter to ensure Response is homogenous (_registered_api_adapter)
      2.5. Run 404 route handler
    3. **When a route is a pre-flight CORS (often not matched)**
      3.1. Exception handlers _(if any exception bubbled up and caught)_
      3.2. Global middlewares _(before, and after on the way back)_
      3.3. Path level middleware _(before, and after on the way back)_
      3.4. Middleware adapter to ensure Response is homogenous (_registered_api_adapter)
      3.5. Return 204 with appropriate CORS headers
    4. **When a route is matched with Data Validation enabled**
      4.1. Exception handlers _(if any exception bubbled up and caught)_
      4.2. Data Validation middleware _(before, and after on the way back)_
      4.3. Global middlewares _(before, and after on the way back)_
      4.4. Path level middleware _(before, and after on the way back)_
      4.5. Middleware adapter to ensure Response is homogenous (_registered_api_adapter)
      4.6. Run actual route

    Parameters
    ----------
    event: dict[str, Any]
        Event
    context: LambdaContext
        Lambda context
    Returns
    -------
    dict
        Returns the dict response
    """
    if isinstance(event, BaseProxyEvent):
        warnings.warn(
            "You don't need to serialize event to Event Source Data Class when using Event Handler; "
            "see issue #1152",
            stacklevel=2,
        )
        event = event.raw_event

    if self._debug:
        print(self._serializer(event))

    # Populate router(s) dependencies without keeping a reference to each registered router
    BaseRouter.current_event = self._to_proxy_event(event)
    BaseRouter.lambda_context = context

    response = self._resolve().build(self.current_event, self._cors)

    # Debug print Processed Middlewares
    if self._debug:
        print("\nProcessed Middlewares:")
        print("======================")
        print("\n".join(self.processed_stack_frames))
        print("======================")

    self.clear_context()

    return response

route

route(
    rule: str,
    method: str | list[str] | tuple[str],
    cors: bool | None = None,
    compress: bool = False,
    cache_control: str | None = None,
    summary: str | None = None,
    description: str | None = None,
    responses: dict[int, OpenAPIResponse] | None = None,
    response_description: str = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
    tags: list[str] | None = None,
    operation_id: str | None = None,
    include_in_schema: bool = True,
    security: list[dict[str, list[str]]] | None = None,
    openapi_extensions: dict[str, Any] | None = None,
    deprecated: bool = False,
    middlewares: list[Callable[..., Any]] | None = None,
) -> Callable[[AnyCallableT], AnyCallableT]

Route decorator includes parameter method

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def route(
    self,
    rule: str,
    method: str | list[str] | tuple[str],
    cors: bool | None = None,
    compress: bool = False,
    cache_control: str | None = None,
    summary: str | None = None,
    description: str | None = None,
    responses: dict[int, OpenAPIResponse] | None = None,
    response_description: str = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
    tags: list[str] | None = None,
    operation_id: str | None = None,
    include_in_schema: bool = True,
    security: list[dict[str, list[str]]] | None = None,
    openapi_extensions: dict[str, Any] | None = None,
    deprecated: bool = False,
    middlewares: list[Callable[..., Any]] | None = None,
) -> Callable[[AnyCallableT], AnyCallableT]:
    """Route decorator includes parameter `method`"""

    def register_resolver(func: AnyCallableT) -> AnyCallableT:
        methods = (method,) if isinstance(method, str) else method
        logger.debug(f"Adding route using rule {rule} and methods: {','.join(m.upper() for m in methods)}")

        cors_enabled = self._cors_enabled if cors is None else cors

        for item in methods:
            _route = Route(
                item,
                rule,
                self._compile_regex(rule),
                func,
                cors_enabled,
                compress,
                cache_control,
                summary,
                description,
                responses,
                response_description,
                tags,
                operation_id,
                include_in_schema,
                security,
                openapi_extensions,
                deprecated,
                middlewares,
            )

            # The more specific route wins.
            # We store dynamic (/studies/{studyid}) and static routes (/studies/fetch) separately.
            # Then attempt a match for static routes before dynamic routes.
            # This ensures that the most specific route is prioritized and processed first (studies/fetch).
            if _route.rule.groups > 0:
                self._dynamic_routes.append(_route)
            else:
                self._static_routes.append(_route)

            self._create_route_key(item, rule)

            if cors_enabled:
                logger.debug(f"Registering method {item.upper()} to Allow Methods in CORS")
                self._cors_methods.add(item.upper())

        return func

    return register_resolver

BaseRouter

Bases: ABC

Base class for Routing

METHOD	DESCRIPTION
append_context	Append key=value data as routing context
clear_context	Resets routing context
delete	Delete route decorator with DELETE method
get	Get route decorator with GET method
head	Head route decorator with HEAD method
patch	Patch route decorator with PATCH method
post	Post route decorator with POST method
put	Put route decorator with PUT method
use	Add one or more global middlewares that run before/after route specific middleware.

append_context

append_context(**additional_context)

Append key=value data as routing context

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def append_context(self, **additional_context):
    """Append key=value data as routing context"""
    self.context.update(**additional_context)

clear_context

clear_context()

Resets routing context

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def clear_context(self):
    """Resets routing context"""
    self.context.clear()

delete

delete(
    rule: str,
    cors: bool | None = None,
    compress: bool = False,
    cache_control: str | None = None,
    summary: str | None = None,
    description: str | None = None,
    responses: dict[int, OpenAPIResponse] | None = None,
    response_description: str = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
    tags: list[str] | None = None,
    operation_id: str | None = None,
    include_in_schema: bool = True,
    security: list[dict[str, list[str]]] | None = None,
    openapi_extensions: dict[str, Any] | None = None,
    deprecated: bool = False,
    middlewares: list[Callable[..., Any]] | None = None,
) -> Callable[[AnyCallableT], AnyCallableT]

Delete route decorator with DELETE method

Examples:

Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator

from aws_lambda_powertools import Tracer
from aws_lambda_powertools.event_handler import APIGatewayRestResolver

tracer = Tracer()
app = APIGatewayRestResolver()

@app.delete("/delete-call")
def simple_delete():
    return {"message": "deleted"}

@tracer.capture_lambda_handler
def lambda_handler(event, context):
    return app.resolve(event, context)

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def delete(
    self,
    rule: str,
    cors: bool | None = None,
    compress: bool = False,
    cache_control: str | None = None,
    summary: str | None = None,
    description: str | None = None,
    responses: dict[int, OpenAPIResponse] | None = None,
    response_description: str = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
    tags: list[str] | None = None,
    operation_id: str | None = None,
    include_in_schema: bool = True,
    security: list[dict[str, list[str]]] | None = None,
    openapi_extensions: dict[str, Any] | None = None,
    deprecated: bool = False,
    middlewares: list[Callable[..., Any]] | None = None,
) -> Callable[[AnyCallableT], AnyCallableT]:
    """Delete route decorator with DELETE `method`

    Examples
    --------
    Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator

    ```python
    from aws_lambda_powertools import Tracer
    from aws_lambda_powertools.event_handler import APIGatewayRestResolver

    tracer = Tracer()
    app = APIGatewayRestResolver()

    @app.delete("/delete-call")
    def simple_delete():
        return {"message": "deleted"}

    @tracer.capture_lambda_handler
    def lambda_handler(event, context):
        return app.resolve(event, context)
    ```
    """
    return self.route(
        rule,
        "DELETE",
        cors,
        compress,
        cache_control,
        summary,
        description,
        responses,
        response_description,
        tags,
        operation_id,
        include_in_schema,
        security,
        openapi_extensions,
        deprecated,
        middlewares,
    )

get

get(
    rule: str,
    cors: bool | None = None,
    compress: bool = False,
    cache_control: str | None = None,
    summary: str | None = None,
    description: str | None = None,
    responses: dict[int, OpenAPIResponse] | None = None,
    response_description: str = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
    tags: list[str] | None = None,
    operation_id: str | None = None,
    include_in_schema: bool = True,
    security: list[dict[str, list[str]]] | None = None,
    openapi_extensions: dict[str, Any] | None = None,
    deprecated: bool = False,
    middlewares: list[Callable[..., Any]] | None = None,
) -> Callable[[AnyCallableT], AnyCallableT]

Get route decorator with GET method

Examples:

Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator

from aws_lambda_powertools import Tracer
from aws_lambda_powertools.event_handler import APIGatewayRestResolver

tracer = Tracer()
app = APIGatewayRestResolver()

@app.get("/get-call")
def simple_get():
    return {"message": "Foo"}

@tracer.capture_lambda_handler
def lambda_handler(event, context):
    return app.resolve(event, context)

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def get(
    self,
    rule: str,
    cors: bool | None = None,
    compress: bool = False,
    cache_control: str | None = None,
    summary: str | None = None,
    description: str | None = None,
    responses: dict[int, OpenAPIResponse] | None = None,
    response_description: str = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
    tags: list[str] | None = None,
    operation_id: str | None = None,
    include_in_schema: bool = True,
    security: list[dict[str, list[str]]] | None = None,
    openapi_extensions: dict[str, Any] | None = None,
    deprecated: bool = False,
    middlewares: list[Callable[..., Any]] | None = None,
) -> Callable[[AnyCallableT], AnyCallableT]:
    """Get route decorator with GET `method`

    Examples
    --------
    Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator

    ```python
    from aws_lambda_powertools import Tracer
    from aws_lambda_powertools.event_handler import APIGatewayRestResolver

    tracer = Tracer()
    app = APIGatewayRestResolver()

    @app.get("/get-call")
    def simple_get():
        return {"message": "Foo"}

    @tracer.capture_lambda_handler
    def lambda_handler(event, context):
        return app.resolve(event, context)
    ```
    """
    return self.route(
        rule,
        "GET",
        cors,
        compress,
        cache_control,
        summary,
        description,
        responses,
        response_description,
        tags,
        operation_id,
        include_in_schema,
        security,
        openapi_extensions,
        deprecated,
        middlewares,
    )

head

head(
    rule: str,
    cors: bool | None = None,
    compress: bool = False,
    cache_control: str | None = None,
    summary: str | None = None,
    description: str | None = None,
    responses: dict[int, OpenAPIResponse] | None = None,
    response_description: str = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
    tags: list[str] | None = None,
    operation_id: str | None = None,
    include_in_schema: bool = True,
    security: list[dict[str, list[str]]] | None = None,
    openapi_extensions: dict[str, Any] | None = None,
    deprecated: bool = False,
    middlewares: list[Callable] | None = None,
) -> Callable[[AnyCallableT], AnyCallableT]

Head route decorator with HEAD method

Examples:

Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator

from aws_lambda_powertools import Tracer
from aws_lambda_powertools.event_handler import APIGatewayRestResolver, Response, content_types

tracer = Tracer()
app = APIGatewayRestResolver()

@app.head("/head-call")
def simple_head():
    return Response(status_code=200,
                    content_type=content_types.APPLICATION_JSON,
                    headers={"Content-Length": "123"})

@tracer.capture_lambda_handler
def lambda_handler(event, context):
    return app.resolve(event, context)

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def head(
    self,
    rule: str,
    cors: bool | None = None,
    compress: bool = False,
    cache_control: str | None = None,
    summary: str | None = None,
    description: str | None = None,
    responses: dict[int, OpenAPIResponse] | None = None,
    response_description: str = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
    tags: list[str] | None = None,
    operation_id: str | None = None,
    include_in_schema: bool = True,
    security: list[dict[str, list[str]]] | None = None,
    openapi_extensions: dict[str, Any] | None = None,
    deprecated: bool = False,
    middlewares: list[Callable] | None = None,
) -> Callable[[AnyCallableT], AnyCallableT]:
    """Head route decorator with HEAD `method`

    Examples
    --------
    Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator

    ```python
    from aws_lambda_powertools import Tracer
    from aws_lambda_powertools.event_handler import APIGatewayRestResolver, Response, content_types

    tracer = Tracer()
    app = APIGatewayRestResolver()

    @app.head("/head-call")
    def simple_head():
        return Response(status_code=200,
                        content_type=content_types.APPLICATION_JSON,
                        headers={"Content-Length": "123"})

    @tracer.capture_lambda_handler
    def lambda_handler(event, context):
        return app.resolve(event, context)
    ```
    """
    return self.route(
        rule,
        "HEAD",
        cors,
        compress,
        cache_control,
        summary,
        description,
        responses,
        response_description,
        tags,
        operation_id,
        include_in_schema,
        security,
        openapi_extensions,
        deprecated,
        middlewares,
    )

patch

patch(
    rule: str,
    cors: bool | None = None,
    compress: bool = False,
    cache_control: str | None = None,
    summary: str | None = None,
    description: str | None = None,
    responses: dict[int, OpenAPIResponse] | None = None,
    response_description: str = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
    tags: list[str] | None = None,
    operation_id: str | None = None,
    include_in_schema: bool = True,
    security: list[dict[str, list[str]]] | None = None,
    openapi_extensions: dict[str, Any] | None = None,
    deprecated: bool = False,
    middlewares: list[Callable] | None = None,
) -> Callable[[AnyCallableT], AnyCallableT]

Patch route decorator with PATCH method

Examples:

Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator

from aws_lambda_powertools import Tracer
from aws_lambda_powertools.event_handler import APIGatewayRestResolver

tracer = Tracer()
app = APIGatewayRestResolver()

@app.patch("/patch-call")
def simple_patch():
    patch_data: dict = app.current_event.json_body
    patch_data["value"] = patched

    return {"message": patch_data}

@tracer.capture_lambda_handler
def lambda_handler(event, context):
    return app.resolve(event, context)

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def patch(
    self,
    rule: str,
    cors: bool | None = None,
    compress: bool = False,
    cache_control: str | None = None,
    summary: str | None = None,
    description: str | None = None,
    responses: dict[int, OpenAPIResponse] | None = None,
    response_description: str = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
    tags: list[str] | None = None,
    operation_id: str | None = None,
    include_in_schema: bool = True,
    security: list[dict[str, list[str]]] | None = None,
    openapi_extensions: dict[str, Any] | None = None,
    deprecated: bool = False,
    middlewares: list[Callable] | None = None,
) -> Callable[[AnyCallableT], AnyCallableT]:
    """Patch route decorator with PATCH `method`

    Examples
    --------
    Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator

    ```python
    from aws_lambda_powertools import Tracer
    from aws_lambda_powertools.event_handler import APIGatewayRestResolver

    tracer = Tracer()
    app = APIGatewayRestResolver()

    @app.patch("/patch-call")
    def simple_patch():
        patch_data: dict = app.current_event.json_body
        patch_data["value"] = patched

        return {"message": patch_data}

    @tracer.capture_lambda_handler
    def lambda_handler(event, context):
        return app.resolve(event, context)
    ```
    """
    return self.route(
        rule,
        "PATCH",
        cors,
        compress,
        cache_control,
        summary,
        description,
        responses,
        response_description,
        tags,
        operation_id,
        include_in_schema,
        security,
        openapi_extensions,
        deprecated,
        middlewares,
    )

post

post(
    rule: str,
    cors: bool | None = None,
    compress: bool = False,
    cache_control: str | None = None,
    summary: str | None = None,
    description: str | None = None,
    responses: dict[int, OpenAPIResponse] | None = None,
    response_description: str = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
    tags: list[str] | None = None,
    operation_id: str | None = None,
    include_in_schema: bool = True,
    security: list[dict[str, list[str]]] | None = None,
    openapi_extensions: dict[str, Any] | None = None,
    deprecated: bool = False,
    middlewares: list[Callable[..., Any]] | None = None,
) -> Callable[[AnyCallableT], AnyCallableT]

Post route decorator with POST method

Examples:

Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator

from aws_lambda_powertools import Tracer
from aws_lambda_powertools.event_handler import APIGatewayRestResolver

tracer = Tracer()
app = APIGatewayRestResolver()

@app.post("/post-call")
def simple_post():
    post_data: dict = app.current_event.json_body
    return {"message": post_data["value"]}

@tracer.capture_lambda_handler
def lambda_handler(event, context):
    return app.resolve(event, context)

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def post(
    self,
    rule: str,
    cors: bool | None = None,
    compress: bool = False,
    cache_control: str | None = None,
    summary: str | None = None,
    description: str | None = None,
    responses: dict[int, OpenAPIResponse] | None = None,
    response_description: str = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
    tags: list[str] | None = None,
    operation_id: str | None = None,
    include_in_schema: bool = True,
    security: list[dict[str, list[str]]] | None = None,
    openapi_extensions: dict[str, Any] | None = None,
    deprecated: bool = False,
    middlewares: list[Callable[..., Any]] | None = None,
) -> Callable[[AnyCallableT], AnyCallableT]:
    """Post route decorator with POST `method`

    Examples
    --------
    Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator

    ```python
    from aws_lambda_powertools import Tracer
    from aws_lambda_powertools.event_handler import APIGatewayRestResolver

    tracer = Tracer()
    app = APIGatewayRestResolver()

    @app.post("/post-call")
    def simple_post():
        post_data: dict = app.current_event.json_body
        return {"message": post_data["value"]}

    @tracer.capture_lambda_handler
    def lambda_handler(event, context):
        return app.resolve(event, context)
    ```
    """
    return self.route(
        rule,
        "POST",
        cors,
        compress,
        cache_control,
        summary,
        description,
        responses,
        response_description,
        tags,
        operation_id,
        include_in_schema,
        security,
        openapi_extensions,
        deprecated,
        middlewares,
    )

put

put(
    rule: str,
    cors: bool | None = None,
    compress: bool = False,
    cache_control: str | None = None,
    summary: str | None = None,
    description: str | None = None,
    responses: dict[int, OpenAPIResponse] | None = None,
    response_description: str = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
    tags: list[str] | None = None,
    operation_id: str | None = None,
    include_in_schema: bool = True,
    security: list[dict[str, list[str]]] | None = None,
    openapi_extensions: dict[str, Any] | None = None,
    deprecated: bool = False,
    middlewares: list[Callable[..., Any]] | None = None,
) -> Callable[[AnyCallableT], AnyCallableT]

Put route decorator with PUT method

Examples:

Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator

from aws_lambda_powertools import Tracer
from aws_lambda_powertools.event_handler import APIGatewayRestResolver

tracer = Tracer()
app = APIGatewayRestResolver()

@app.put("/put-call")
def simple_put():
    put_data: dict = app.current_event.json_body
    return {"message": put_data["value"]}

@tracer.capture_lambda_handler
def lambda_handler(event, context):
    return app.resolve(event, context)

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def put(
    self,
    rule: str,
    cors: bool | None = None,
    compress: bool = False,
    cache_control: str | None = None,
    summary: str | None = None,
    description: str | None = None,
    responses: dict[int, OpenAPIResponse] | None = None,
    response_description: str = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
    tags: list[str] | None = None,
    operation_id: str | None = None,
    include_in_schema: bool = True,
    security: list[dict[str, list[str]]] | None = None,
    openapi_extensions: dict[str, Any] | None = None,
    deprecated: bool = False,
    middlewares: list[Callable[..., Any]] | None = None,
) -> Callable[[AnyCallableT], AnyCallableT]:
    """Put route decorator with PUT `method`

    Examples
    --------
    Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator

    ```python
    from aws_lambda_powertools import Tracer
    from aws_lambda_powertools.event_handler import APIGatewayRestResolver

    tracer = Tracer()
    app = APIGatewayRestResolver()

    @app.put("/put-call")
    def simple_put():
        put_data: dict = app.current_event.json_body
        return {"message": put_data["value"]}

    @tracer.capture_lambda_handler
    def lambda_handler(event, context):
        return app.resolve(event, context)
    ```
    """
    return self.route(
        rule,
        "PUT",
        cors,
        compress,
        cache_control,
        summary,
        description,
        responses,
        response_description,
        tags,
        operation_id,
        include_in_schema,
        security,
        openapi_extensions,
        deprecated,
        middlewares,
    )

use

use(middlewares: list[Callable[..., Response]]) -> None

Add one or more global middlewares that run before/after route specific middleware.

NOTE: Middlewares are called in insertion order.

PARAMETER	DESCRIPTION
middlewares	List of global middlewares to be used TYPE: list[Callable[..., Response]]

Examples:

Add middlewares to be used for every request processed by the Router.

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

logger = Logger()
app = APIGatewayRestResolver()

def log_request_response(app: APIGatewayRestResolver, next_middleware: NextMiddleware) -> Response:
    logger.info("Incoming request", path=app.current_event.path, request=app.current_event.raw_event)

    result = next_middleware(app)
    logger.info("Response received", response=result.__dict__)

    return result

app.use(middlewares=[log_request_response])


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

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def use(self, middlewares: list[Callable[..., Response]]) -> None:
    """
    Add one or more global middlewares that run before/after route specific middleware.

    NOTE: Middlewares are called in insertion order.

    Parameters
    ----------
    middlewares: list[Callable[..., Response]]
        List of global middlewares to be used

    Examples
    --------

    Add middlewares to be used for every request processed by the Router.

    ```python
    from aws_lambda_powertools import Logger
    from aws_lambda_powertools.event_handler import APIGatewayRestResolver, Response
    from aws_lambda_powertools.event_handler.middlewares import NextMiddleware

    logger = Logger()
    app = APIGatewayRestResolver()

    def log_request_response(app: APIGatewayRestResolver, next_middleware: NextMiddleware) -> Response:
        logger.info("Incoming request", path=app.current_event.path, request=app.current_event.raw_event)

        result = next_middleware(app)
        logger.info("Response received", response=result.__dict__)

        return result

    app.use(middlewares=[log_request_response])


    def lambda_handler(event, context):
        return app.resolve(event, context)
    ```
    """
    self._router_middlewares = self._router_middlewares + middlewares

CORSConfig

CORSConfig(
    allow_origin: str = "*",
    extra_origins: list[str] | None = None,
    allow_headers: list[str] | None = None,
    expose_headers: list[str] | None = None,
    max_age: int | None = None,
    allow_credentials: bool = False,
)

CORS Config

Examples:

Simple CORS example using the default permissive CORS, note that this should only be used during early prototyping.

from aws_lambda_powertools.event_handler.api_gateway import (
    APIGatewayRestResolver, CORSConfig
)

app = APIGatewayRestResolver(cors=CORSConfig())

@app.get("/my/path")
def with_cors():
    return {"message": "Foo"}

Using a custom CORSConfig where with_cors used the custom provided CORSConfig and without_cors do not include any CORS headers.

from aws_lambda_powertools.event_handler.api_gateway import (
    APIGatewayRestResolver, CORSConfig
)

cors_config = CORSConfig(
    allow_origin="https://wwww.example.com/",
    extra_origins=["https://dev.example.com/"],
    expose_headers=["x-exposed-response-header"],
    allow_headers=["x-custom-request-header"],
    max_age=100,
    allow_credentials=True,
)
app = APIGatewayRestResolver(cors=cors_config)

@app.get("/my/path")
def with_cors():
    return {"message": "Foo"}

@app.get("/another-one", cors=False)
def without_cors():
    return {"message": "Foo"}

METHOD	DESCRIPTION
build_allow_methods	Build sorted comma delimited methods for Access-Control-Allow-Methods header
to_dict	Builds the configured Access-Control http headers

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def __init__(
    self,
    allow_origin: str = "*",
    extra_origins: list[str] | None = None,
    allow_headers: list[str] | None = None,
    expose_headers: list[str] | None = None,
    max_age: int | None = None,
    allow_credentials: bool = False,
):
    """
    Parameters
    ----------
    allow_origin: str
        The value of the `Access-Control-Allow-Origin` to send in the response. Defaults to "*", but should
        only be used during development.
    extra_origins: list[str] | None
        The list of additional allowed origins.
    allow_headers: list[str] | None
        The list of additional allowed headers. This list is added to list of
        built-in allowed headers: `Authorization`, `Content-Type`, `X-Amz-Date`,
        `X-Api-Key`, `X-Amz-Security-Token`.
    expose_headers: list[str] | None
        A list of values to return for the Access-Control-Expose-Headers
    max_age: int | None
        The value for the `Access-Control-Max-Age`
    allow_credentials: bool
        A boolean value that sets the value of `Access-Control-Allow-Credentials`
    """

    self._allowed_origins = [allow_origin]

    if extra_origins:
        self._allowed_origins.extend(extra_origins)

    self.allow_headers = set(self._REQUIRED_HEADERS + (allow_headers or []))
    self.expose_headers = expose_headers or []
    self.max_age = max_age
    self.allow_credentials = allow_credentials

build_allow_methods staticmethod

build_allow_methods(methods: set[str]) -> str

Build sorted comma delimited methods for Access-Control-Allow-Methods header

PARAMETER	DESCRIPTION
methods	Set of HTTP Methods TYPE: set[str]

RETURNS	DESCRIPTION
set[str]	Formatted string with all HTTP Methods allowed for CORS e.g., GET, OPTIONS

Source code in aws_lambda_powertools/event_handler/api_gateway.py

@staticmethod
def build_allow_methods(methods: set[str]) -> str:
    """Build sorted comma delimited methods for Access-Control-Allow-Methods header

    Parameters
    ----------
    methods : set[str]
        Set of HTTP Methods

    Returns
    -------
    set[str]
        Formatted string with all HTTP Methods allowed for CORS e.g., `GET, OPTIONS`

    """
    return ",".join(sorted(methods))

to_dict

to_dict(origin: str | None) -> dict[str, str]

Builds the configured Access-Control http headers

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def to_dict(self, origin: str | None) -> dict[str, str]:
    """Builds the configured Access-Control http headers"""

    # If there's no Origin, don't add any CORS headers
    if not origin:
        return {}

    # If the origin doesn't match any of the allowed origins, and we don't allow all origins ("*"),
    # don't add any CORS headers
    if origin not in self._allowed_origins and "*" not in self._allowed_origins:
        return {}

    # The origin matched an allowed origin, so return the CORS headers
    headers = {
        "Access-Control-Allow-Origin": origin,
        "Access-Control-Allow-Headers": CORSConfig.build_allow_methods(self.allow_headers),
    }

    if self.expose_headers:
        headers["Access-Control-Expose-Headers"] = ",".join(self.expose_headers)
    if self.max_age is not None:
        headers["Access-Control-Max-Age"] = str(self.max_age)
    if origin != "*" and self.allow_credentials is True:
        headers["Access-Control-Allow-Credentials"] = "true"
    return headers

MiddlewareFrame

MiddlewareFrame(
    current_middleware: Callable[..., Any],
    next_middleware: Callable[..., Any],
)

Creates a Middle Stack Wrapper instance to be used as a "Frame" in the overall stack of middleware functions. Each instance contains the current middleware and the next middleware function to be called in the stack.

In this way the middleware stack is constructed in a recursive fashion, with each middleware calling the next as a simple function call. The actual Python call-stack will contain each MiddlewareStackWrapper "Frame", meaning any Middleware function can cause the entire Middleware call chain to be exited early (short-circuited) by raising an exception or by simply returning early with a custom Response. The decision to short-circuit the middleware chain is at the user's discretion but instantly available due to the Wrapped nature of the callable constructs in the Middleware stack and each Middleware function having complete control over whether the "Next" handler in the stack is called or not.

PARAMETER	DESCRIPTION
current_middleware	The current middleware function to be called as a request is processed. TYPE: Callable
next_middleware	The next middleware in the middleware stack. TYPE: Callable

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def __init__(
    self,
    current_middleware: Callable[..., Any],
    next_middleware: Callable[..., Any],
) -> None:
    self.current_middleware: Callable[..., Any] = current_middleware
    self.next_middleware: Callable[..., Any] = next_middleware
    self._next_middleware_name = next_middleware.__name__

ProxyEventType

Bases: Enum

An enumerations of the supported proxy event types.

Response

Response(
    status_code: int,
    content_type: str | None = None,
    body: ResponseT | None = None,
    headers: Mapping[str, str | list[str]] | None = None,
    cookies: list[Cookie] | None = None,
    compress: bool | None = None,
)

Bases: Generic[ResponseT]

Response data class that provides greater control over what is returned from the proxy event

METHOD	DESCRIPTION
is_json	Returns True if the response is JSON, based on the Content-Type.

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def __init__(
    self,
    status_code: int,
    content_type: str | None = None,
    body: ResponseT | None = None,
    headers: Mapping[str, str | list[str]] | None = None,
    cookies: list[Cookie] | None = None,
    compress: bool | None = None,
):
    """

    Parameters
    ----------
    status_code: int
        Http status code, example 200
    content_type: str
        Optionally set the Content-Type header, example "application/json". Note this will be merged into any
        provided http headers
    body: str | bytes | None
        Optionally set the response body. Note: bytes body will be automatically base64 encoded
    headers: Mapping[str, str | list[str]]
        Optionally set specific http headers. Setting "Content-Type" here would override the `content_type` value.
    cookies: list[Cookie]
        Optionally set cookies.
    """
    self.status_code = status_code
    self.body = body
    self.base64_encoded = False
    self.headers: dict[str, str | list[str]] = dict(headers) if headers else {}
    self.cookies = cookies or []
    self.compress = compress
    self.content_type = content_type
    if content_type:
        self.headers.setdefault("Content-Type", content_type)

is_json

is_json() -> bool

Returns True if the response is JSON, based on the Content-Type.

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def is_json(self) -> bool:
    """
    Returns True if the response is JSON, based on the Content-Type.
    """
    content_type = self.headers.get("Content-Type", "")
    if isinstance(content_type, list):
        content_type = content_type[0]
    return content_type.startswith("application/json")

ResponseBuilder

ResponseBuilder(
    response: Response,
    serializer: Callable[[Any], str] = partial(
        json.dumps, separators=(",", ":"), cls=Encoder
    ),
    route: Route | None = None,
)

Bases: Generic[ResponseEventT]

Internally used Response builder

METHOD	DESCRIPTION
build	Build the full response dict to be returned by the lambda

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def __init__(
    self,
    response: Response,
    serializer: Callable[[Any], str] = partial(json.dumps, separators=(",", ":"), cls=Encoder),
    route: Route | None = None,
):
    self.response = response
    self.serializer = serializer
    self.route = route

build

build(
    event: ResponseEventT, cors: CORSConfig | None = None
) -> dict[str, Any]

Build the full response dict to be returned by the lambda

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def build(self, event: ResponseEventT, cors: CORSConfig | None = None) -> dict[str, Any]:
    """Build the full response dict to be returned by the lambda"""

    # We only apply the serializer when the content type is JSON and the
    # body is not a str, to avoid double encoding
    if self.response.is_json() and not isinstance(self.response.body, str):
        self.response.body = self.serializer(self.response.body)

    self._route(event, cors)

    if isinstance(self.response.body, bytes):
        logger.debug("Encoding bytes response with base64")
        self.response.base64_encoded = True
        self.response.body = base64.b64encode(self.response.body).decode()

    return {
        "statusCode": self.response.status_code,
        "body": self.response.body,
        "isBase64Encoded": self.response.base64_encoded,
        **event.header_serializer().serialize(headers=self.response.headers, cookies=self.response.cookies),
    }

Route

Route(
    method: str,
    path: str,
    rule: Pattern,
    func: Callable,
    cors: bool,
    compress: bool,
    cache_control: str | None = None,
    summary: str | None = None,
    description: str | None = None,
    responses: dict[int, OpenAPIResponse] | None = None,
    response_description: str | None = None,
    tags: list[str] | None = None,
    operation_id: str | None = None,
    include_in_schema: bool = True,
    security: list[dict[str, list[str]]] | None = None,
    openapi_extensions: dict[str, Any] | None = None,
    deprecated: bool = False,
    middlewares: (
        list[Callable[..., Response]] | None
    ) = None,
)

Internally used Route Configuration

PARAMETER	DESCRIPTION
method	The HTTP method, example "GET" TYPE: str
path	The path of the route TYPE: str
rule	The route rule, example "/my/path" TYPE: Pattern
func	The route handler function TYPE: Callable
cors	Whether or not to enable CORS for this route TYPE: bool
compress	Whether or not to enable gzip compression for this route TYPE: bool
cache_control	The cache control header value, example "max-age=3600" TYPE: str | None DEFAULT: None
summary	The OpenAPI summary for this route TYPE: str | None DEFAULT: None
description	The OpenAPI description for this route TYPE: str | None DEFAULT: None
responses	The OpenAPI responses for this route TYPE: dict[int, OpenAPIResponse] | None DEFAULT: None
response_description	The OpenAPI response description for this route TYPE: str | None DEFAULT: None
tags	The list of OpenAPI tags to be used for this route TYPE: list[str] | None DEFAULT: None
operation_id	The OpenAPI operationId for this route TYPE: str | None DEFAULT: None
include_in_schema	Whether or not to include this route in the OpenAPI schema TYPE: bool DEFAULT: True
security	The OpenAPI security for this route TYPE: list[dict[str, list[str]]] | None DEFAULT: None
openapi_extensions	Additional OpenAPI extensions as a dictionary. TYPE: dict[str, Any] | None DEFAULT: None
deprecated	Whether or not to mark this route as deprecated in the OpenAPI schema TYPE: bool DEFAULT: False
middlewares	The list of route middlewares to be called in order. TYPE: list[Callable[..., Response]] | None DEFAULT: None

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def __init__(
    self,
    method: str,
    path: str,
    rule: Pattern,
    func: Callable,
    cors: bool,
    compress: bool,
    cache_control: str | None = None,
    summary: str | None = None,
    description: str | None = None,
    responses: dict[int, OpenAPIResponse] | None = None,
    response_description: str | None = None,
    tags: list[str] | None = None,
    operation_id: str | None = None,
    include_in_schema: bool = True,
    security: list[dict[str, list[str]]] | None = None,
    openapi_extensions: dict[str, Any] | None = None,
    deprecated: bool = False,
    middlewares: list[Callable[..., Response]] | None = None,
):
    """
    Internally used Route Configuration

    Parameters
    ----------
    method: str
        The HTTP method, example "GET"
    path: str
        The path of the route
    rule: Pattern
        The route rule, example "/my/path"
    func: Callable
        The route handler function
    cors: bool
        Whether or not to enable CORS for this route
    compress: bool
        Whether or not to enable gzip compression for this route
    cache_control: str | None
        The cache control header value, example "max-age=3600"
    summary: str | None
        The OpenAPI summary for this route
    description: str | None
        The OpenAPI description for this route
    responses: dict[int, OpenAPIResponse] | None
        The OpenAPI responses for this route
    response_description: str | None
        The OpenAPI response description for this route
    tags: list[str] | None
        The list of OpenAPI tags to be used for this route
    operation_id: str | None
        The OpenAPI operationId for this route
    include_in_schema: bool
        Whether or not to include this route in the OpenAPI schema
    security: list[dict[str, list[str]]], optional
        The OpenAPI security for this route
    openapi_extensions: dict[str, Any], optional
        Additional OpenAPI extensions as a dictionary.
    deprecated: bool
        Whether or not to mark this route as deprecated in the OpenAPI schema
    middlewares: list[Callable[..., Response]] | None
        The list of route middlewares to be called in order.
    """
    self.method = method.upper()
    self.path = "/" if path.strip() == "" else path

    # OpenAPI spec only understands paths with { }. So we'll have to convert Powertools' < >.
    # https://swagger.io/specification/#path-templating
    self.openapi_path = re.sub(r"<(.*?)>", lambda m: f"{{{''.join(m.group(1))}}}", self.path)

    self.rule = rule
    self.func = func
    self._middleware_stack = func
    self.cors = cors
    self.compress = compress
    self.cache_control = cache_control
    self.summary = summary
    self.description = description
    self.responses = responses
    self.response_description = response_description
    self.tags = tags or []
    self.include_in_schema = include_in_schema
    self.security = security
    self.openapi_extensions = openapi_extensions
    self.middlewares = middlewares or []
    self.operation_id = operation_id or self._generate_operation_id()
    self.deprecated = deprecated

    # _middleware_stack_built is used to ensure the middleware stack is only built once.
    self._middleware_stack_built = False

    # _dependant is used to cache the dependant model for the handler function
    self._dependant: Dependant | None = None

    # _body_field is used to cache the dependant model for the body field
    self._body_field: ModelField | None = None

Router

Router()

Bases: BaseRouter

Router helper class to allow splitting ApiGatewayResolver into multiple files

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def __init__(self):
    self._routes: dict[tuple, Callable] = {}
    self._routes_with_middleware: dict[tuple, list[Callable]] = {}
    self.api_resolver: BaseRouter | None = None
    self.context = {}  # early init as customers might add context before event resolution
    self._exception_handlers: dict[type, Callable] = {}