Skip to content

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,
    response_validation_error_http_code: (
        HTTPStatus | int | None
    ) = None,
)

Bases: ApiGatewayResolver

Amazon Application Load Balancer (ALB) resolver

Source code in aws_lambda_powertools/event_handler/api_gateway.py
2869
2870
2871
2872
2873
2874
2875
2876
2877
2878
2879
2880
2881
2882
2883
2884
2885
2886
2887
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,
    response_validation_error_http_code: HTTPStatus | int | None = None,
):
    """Amazon Application Load Balancer (ALB) resolver"""
    super().__init__(
        ProxyEventType.ALBEvent,
        cors,
        debug,
        serializer,
        strip_prefixes,
        enable_validation,
        response_validation_error_http_code,
    )

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,
    response_validation_error_http_code: (
        HTTPStatus | int | None
    ) = None,
)

Bases: ApiGatewayResolver

Amazon API Gateway HTTP API v2 payload resolver

Source code in aws_lambda_powertools/event_handler/api_gateway.py
2830
2831
2832
2833
2834
2835
2836
2837
2838
2839
2840
2841
2842
2843
2844
2845
2846
2847
2848
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,
    response_validation_error_http_code: HTTPStatus | int | None = None,
):
    """Amazon API Gateway HTTP API v2 payload resolver"""
    super().__init__(
        ProxyEventType.APIGatewayProxyEventV2,
        cors,
        debug,
        serializer,
        strip_prefixes,
        enable_validation,
        response_validation_error_http_code,
    )

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,
    response_validation_error_http_code: (
        HTTPStatus | int | None
    ) = None,
)

Bases: ApiGatewayResolver

Amazon API Gateway REST and HTTP API v1 payload resolver

Source code in aws_lambda_powertools/event_handler/api_gateway.py
2746
2747
2748
2749
2750
2751
2752
2753
2754
2755
2756
2757
2758
2759
2760
2761
2762
2763
2764
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,
    response_validation_error_http_code: HTTPStatus | int | None = None,
):
    """Amazon API Gateway REST and HTTP API v1 payload resolver"""
    super().__init__(
        ProxyEventType.APIGatewayProxyEvent,
        cors,
        debug,
        serializer,
        strip_prefixes,
        enable_validation,
        response_validation_error_http_code,
    )

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,
    response_validation_error_http_code: (
        HTTPStatus | int | None
    ) = None,
)

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
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)

response_validation_error_http_code Sets the returned status code if response is not validated. enable_validation must be True.

METHOD DESCRIPTION
configure_openapi

Configure OpenAPI specification settings for the API.

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
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
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,
    response_validation_error_http_code: HTTPStatus | int | None = None,
):
    """
    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.
    response_validation_error_http_code
        Sets the returned status code if response is not validated. enable_validation must be True.
    """
    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]
    self.openapi_config = OpenAPIConfig()  # starting an empty dataclass
    self._has_response_validation_error = response_validation_error_http_code is not None
    self._response_validation_error_http_code = self._validate_response_validation_error_http_code(
        response_validation_error_http_code,
        enable_validation,
    )

    # 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,
                    has_response_validation_error=self._has_response_validation_error,
                ),
            ],
        )

configure_openapi

configure_openapi(
    title: str = DEFAULT_OPENAPI_TITLE,
    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,
)

Configure OpenAPI specification settings for the API.

Sets up the OpenAPI documentation configuration that can be later used when enabling Swagger UI or generating OpenAPI specifications.

PARAMETER DESCRIPTION
title

The title of the application.

TYPE: str DEFAULT: DEFAULT_OPENAPI_TITLE

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

Example

api.configure_openapi( ... title="My API", ... version="1.0.0", ... description="API for managing resources", ... contact=Contact( ... name="API Support", ... email="support@example.com" ... ) ... )

See Also

enable_swagger : Method to enable Swagger UI using these configurations OpenAPIConfig : Data class containing all OpenAPI configuration options

Source code in aws_lambda_powertools/event_handler/api_gateway.py
1874
1875
1876
1877
1878
1879
1880
1881
1882
1883
1884
1885
1886
1887
1888
1889
1890
1891
1892
1893
1894
1895
1896
1897
1898
1899
1900
1901
1902
1903
1904
1905
1906
1907
1908
1909
1910
1911
1912
1913
1914
1915
1916
1917
1918
1919
1920
1921
1922
1923
1924
1925
1926
1927
1928
1929
1930
1931
1932
1933
1934
1935
1936
1937
1938
1939
1940
1941
1942
1943
1944
1945
1946
1947
1948
1949
1950
1951
1952
1953
1954
1955
def configure_openapi(
    self,
    title: str = DEFAULT_OPENAPI_TITLE,
    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,
):
    """Configure OpenAPI specification settings for the API.

    Sets up the OpenAPI documentation configuration that can be later used
    when enabling Swagger UI or generating OpenAPI specifications.

    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.

    Example
    --------
    >>> api.configure_openapi(
    ...     title="My API",
    ...     version="1.0.0",
    ...     description="API for managing resources",
    ...     contact=Contact(
    ...         name="API Support",
    ...         email="support@example.com"
    ...     )
    ... )

    See Also
    --------
    enable_swagger : Method to enable Swagger UI using these configurations
    OpenAPIConfig : Data class containing all OpenAPI configuration options
    """
    self.openapi_config = OpenAPIConfig(
        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,
    )

enable_swagger

enable_swagger(
    *,
    path: str = "/swagger",
    title: str = DEFAULT_OPENAPI_TITLE,
    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: DEFAULT_OPENAPI_TITLE

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
1957
1958
1959
1960
1961
1962
1963
1964
1965
1966
1967
1968
1969
1970
1971
1972
1973
1974
1975
1976
1977
1978
1979
1980
1981
1982
1983
1984
1985
1986
1987
1988
1989
1990
1991
1992
1993
1994
1995
1996
1997
1998
1999
2000
2001
2002
2003
2004
2005
2006
2007
2008
2009
2010
2011
2012
2013
2014
2015
2016
2017
2018
2019
2020
2021
2022
2023
2024
2025
2026
2027
2028
2029
2030
2031
2032
2033
2034
2035
2036
2037
2038
2039
2040
2041
2042
2043
2044
2045
2046
2047
2048
2049
2050
2051
2052
2053
2054
2055
2056
2057
2058
2059
2060
2061
2062
2063
2064
2065
2066
2067
2068
2069
2070
2071
2072
2073
2074
2075
2076
2077
2078
2079
2080
2081
2082
2083
2084
2085
2086
2087
2088
2089
2090
2091
2092
2093
2094
2095
2096
2097
2098
2099
2100
2101
2102
2103
2104
2105
2106
2107
2108
2109
def enable_swagger(
    self,
    *,
    path: str = "/swagger",
    title: str = DEFAULT_OPENAPI_TITLE,
    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 = DEFAULT_OPENAPI_TITLE,
    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: DEFAULT_OPENAPI_TITLE

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
1796
1797
1798
1799
1800
1801
1802
1803
1804
1805
1806
1807
1808
1809
1810
1811
1812
1813
1814
1815
1816
1817
1818
1819
1820
1821
1822
1823
1824
1825
1826
1827
1828
1829
1830
1831
1832
1833
1834
1835
1836
1837
1838
1839
1840
1841
1842
1843
1844
1845
1846
1847
1848
1849
1850
1851
1852
1853
1854
1855
1856
1857
1858
1859
1860
1861
1862
1863
1864
1865
1866
1867
1868
1869
1870
1871
1872
def get_openapi_json_schema(
    self,
    *,
    title: str = DEFAULT_OPENAPI_TITLE,
    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 = DEFAULT_OPENAPI_TITLE,
    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: DEFAULT_OPENAPI_TITLE

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
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
1681
1682
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707
1708
1709
1710
1711
1712
1713
1714
1715
1716
1717
1718
1719
1720
1721
1722
1723
1724
1725
1726
1727
1728
1729
1730
1731
1732
1733
1734
1735
1736
1737
1738
1739
1740
1741
1742
1743
1744
1745
1746
1747
1748
1749
1750
1751
1752
1753
1754
1755
1756
1757
1758
1759
def get_openapi_schema(
    self,
    *,
    title: str = DEFAULT_OPENAPI_TITLE,
    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.
    """

    # DEPRECATION: Will be removed in v4.0.0. Use configure_api() instead.
    # Maintained for backwards compatibility.
    # See: https://github.com/aws-powertools/powertools-lambda-python/issues/6122
    if title == DEFAULT_OPENAPI_TITLE and self.openapi_config.title:
        title = self.openapi_config.title

    if version == DEFAULT_API_VERSION and self.openapi_config.version:
        version = self.openapi_config.version

    if openapi_version == DEFAULT_OPENAPI_VERSION and self.openapi_config.openapi_version:
        openapi_version = self.openapi_config.openapi_version

    summary = summary or self.openapi_config.summary
    description = description or self.openapi_config.description
    tags = tags or self.openapi_config.tags
    servers = servers or self.openapi_config.servers
    terms_of_service = terms_of_service or self.openapi_config.terms_of_service
    contact = contact or self.openapi_config.contact
    license_info = license_info or self.openapi_config.license_info
    security_schemes = security_schemes or self.openapi_config.security_schemes
    security = security or self.openapi_config.security
    openapi_extensions = openapi_extensions or self.openapi_config.openapi_extensions

    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
2577
2578
2579
2580
2581
2582
2583
2584
2585
2586
2587
2588
2589
2590
2591
2592
2593
2594
2595
2596
2597
2598
2599
2600
2601
2602
2603
2604
2605
2606
2607
2608
2609
2610
2611
2612
2613
2614
2615
2616
2617
2618
2619
2620
2621
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
2179
2180
2181
2182
2183
2184
2185
2186
2187
2188
2189
2190
2191
2192
2193
2194
2195
2196
2197
2198
2199
2200
2201
2202
2203
2204
2205
2206
2207
2208
2209
2210
2211
2212
2213
2214
2215
2216
2217
2218
2219
2220
2221
2222
2223
2224
2225
2226
2227
2228
2229
2230
2231
2232
2233
2234
2235
2236
2237
2238
2239
2240
2241
2242
2243
2244
2245
2246
2247
2248
2249
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
2111
2112
2113
2114
2115
2116
2117
2118
2119
2120
2121
2122
2123
2124
2125
2126
2127
2128
2129
2130
2131
2132
2133
2134
2135
2136
2137
2138
2139
2140
2141
2142
2143
2144
2145
2146
2147
2148
2149
2150
2151
2152
2153
2154
2155
2156
2157
2158
2159
2160
2161
2162
2163
2164
2165
2166
2167
2168
2169
2170
2171
2172
2173
2174
2175
2176
2177
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
1363
1364
1365
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
1367
1368
1369
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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
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
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
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
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
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
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
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
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
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
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
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
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
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.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
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
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
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.

1
2
3
4
5
6
7
8
9
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.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
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
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
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
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
@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
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
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
1395
1396
1397
1398
1399
1400
1401
1402
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
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
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
290
291
292
293
294
295
296
297
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
814
815
816
817
818
819
820
821
822
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
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
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
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
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
2661
2662
2663
2664
2665
2666
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] = {}