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
`BedrockResponse`	Contains the response body, status code, content type, and optional attributes
`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

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

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

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

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

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.exception_handler_manager = ExceptionHandlerManager()
    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" ... ) ... )

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

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=_DEFAULT_CONTENT_TYPE,
                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

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

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 pydantic.json_schema import GenerateJsonSchema

    from aws_lambda_powertools.event_handler.openapi.compat import (
        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 = self._add_resolver_response_validation_error_response_to_route(result)
            if path:
                paths.setdefault(route.openapi_path, {}).update(path)
            if path_definitions:
                definitions.update(path_definitions)

    if definitions:
        components["schemas"] = self._generate_schemas(definitions)
    if security_schemes:
        components["securitySchemes"] = security_schemes
    if components:
        output["components"] = components
    if tags:
        output["tags"] = [Tag(name=tag) if isinstance(tag, str) else tag for tag in tags]

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

    return OpenAPI(**output)

include_router ¶

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

Adds all routes and context defined in a router

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

Source code in aws_lambda_powertools/event_handler/api_gateway.py

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

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

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

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

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

    logger.debug("Appending Router exception_handler into App exception_handler.")
    self.exception_handler_manager.update_exception_handlers(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__):

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
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
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
When a route is matched with Data Validation enabled 4.1. Exception handlers (if any exception bubbled up and caught) 4.2. Data Validation middleware (before, and after on the way back) 4.3. Global middlewares (before, and after on the way back) 4.4. Path level middleware (before, and after on the way back) 4.5. Middleware adapter to ensure Response is homogenous (_registered_api_adapter) 4.6. Run actual route

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

RETURNS	DESCRIPTION
`dict`	Returns the dict response

Source code in aws_lambda_powertools/event_handler/api_gateway.py

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

    ## Internals

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

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

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

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

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

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

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

    self.clear_context()

    return response

route ¶

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

Route decorator includes parameter method

Source code in aws_lambda_powertools/event_handler/api_gateway.py

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

    custom_response_validation_http_code = self._validate_route_response_validation_error_http_code(
        custom_response_validation_http_code,
    )

    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,
                custom_response_validation_http_code,
                middlewares,
            )

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

            self._create_route_key(item, rule)

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

        return func

    return register_resolver

BaseRouter ¶

Bases: ABC

Base class for Routing

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

append_context ¶

append_context(**additional_context)

Append key=value data as routing context

Source code in aws_lambda_powertools/event_handler/api_gateway.py

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

clear_context ¶

clear_context()

Resets routing context

Source code in aws_lambda_powertools/event_handler/api_gateway.py

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

delete ¶

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

Delete route decorator with DELETE method

Examples:

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

from aws_lambda_powertools import Tracer
from aws_lambda_powertools.event_handler import APIGatewayRestResolver

tracer = Tracer()
app = APIGatewayRestResolver()

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

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

Source code in aws_lambda_powertools/event_handler/api_gateway.py

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

Get route decorator with GET method

Examples:

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

from aws_lambda_powertools import Tracer
from aws_lambda_powertools.event_handler import APIGatewayRestResolver

tracer = Tracer()
app = APIGatewayRestResolver()

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

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

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def get(
    self,
    rule: str,
    cors: bool | None = None,
    compress: bool = False,
    cache_control: str | None = None,
    summary: str | None = None,
    description: str | None = None,
    responses: dict[int, OpenAPIResponse] | None = None,
    response_description: str = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
    tags: list[str] | None = None,
    operation_id: str | None = None,
    include_in_schema: bool = True,
    security: list[dict[str, list[str]]] | None = None,
    openapi_extensions: dict[str, Any] | None = None,
    deprecated: bool = False,
    custom_response_validation_http_code: int | HTTPStatus | None = None,
    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,
        custom_response_validation_http_code,
        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,
    custom_response_validation_http_code: int
    | HTTPStatus
    | None = None,
    middlewares: list[Callable] | None = None,
) -> Callable[[AnyCallableT], AnyCallableT]

Head route decorator with HEAD method

Examples:

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

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

tracer = Tracer()
app = APIGatewayRestResolver()

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

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

Source code in aws_lambda_powertools/event_handler/api_gateway.py

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

Patch route decorator with PATCH method

Examples:

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

from aws_lambda_powertools import Tracer
from aws_lambda_powertools.event_handler import APIGatewayRestResolver

tracer = Tracer()
app = APIGatewayRestResolver()

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

    return {"message": patch_data}

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

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def patch(
    self,
    rule: str,
    cors: bool | None = None,
    compress: bool = False,
    cache_control: str | None = None,
    summary: str | None = None,
    description: str | None = None,
    responses: dict[int, OpenAPIResponse] | None = None,
    response_description: str = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
    tags: list[str] | None = None,
    operation_id: str | None = None,
    include_in_schema: bool = True,
    security: list[dict[str, list[str]]] | None = None,
    openapi_extensions: dict[str, Any] | None = None,
    deprecated: bool = False,
    custom_response_validation_http_code: int | HTTPStatus | None = None,
    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,
        custom_response_validation_http_code,
        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,
    custom_response_validation_http_code: int
    | HTTPStatus
    | None = None,
    middlewares: list[Callable[..., Any]] | None = None,
) -> Callable[[AnyCallableT], AnyCallableT]

Post route decorator with POST method

Examples:

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

from aws_lambda_powertools import Tracer
from aws_lambda_powertools.event_handler import APIGatewayRestResolver

tracer = Tracer()
app = APIGatewayRestResolver()

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

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

Source code in aws_lambda_powertools/event_handler/api_gateway.py

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

Put route decorator with PUT method

Examples:

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

from aws_lambda_powertools import Tracer
from aws_lambda_powertools.event_handler import APIGatewayRestResolver

tracer = Tracer()
app = APIGatewayRestResolver()

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

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

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def put(
    self,
    rule: str,
    cors: bool | None = None,
    compress: bool = False,
    cache_control: str | None = None,
    summary: str | None = None,
    description: str | None = None,
    responses: dict[int, OpenAPIResponse] | None = None,
    response_description: str = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
    tags: list[str] | None = None,
    operation_id: str | None = None,
    include_in_schema: bool = True,
    security: list[dict[str, list[str]]] | None = None,
    openapi_extensions: dict[str, Any] | None = None,
    deprecated: bool = False,
    custom_response_validation_http_code: int | HTTPStatus | None = None,
    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,
        custom_response_validation_http_code,
        middlewares,
    )

use ¶

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

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

NOTE: Middlewares are called in insertion order.

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

Examples:

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

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

logger = Logger()
app = APIGatewayRestResolver()

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

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

    return result

app.use(middlewares=[log_request_response])


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

Source code in aws_lambda_powertools/event_handler/api_gateway.py

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

    NOTE: Middlewares are called in insertion order.

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

    Examples
    --------

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

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

    logger = Logger()
    app = APIGatewayRestResolver()

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

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

        return result

    app.use(middlewares=[log_request_response])


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

BedrockResponse ¶

BedrockResponse(
    body: Any = None,
    status_code: int = 200,
    content_type: str = _DEFAULT_CONTENT_TYPE,
    session_attributes: dict[str, Any] | None = None,
    prompt_session_attributes: dict[str, Any] | None = None,
    knowledge_bases_configuration: list[dict[str, Any]]
    | None = None,
)

Bases: Generic[ResponseT]

Contains the response body, status code, content type, and optional attributes for session management and knowledge base configuration.

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

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def __init__(
    self,
    body: Any = None,
    status_code: int = 200,
    content_type: str = _DEFAULT_CONTENT_TYPE,
    session_attributes: dict[str, Any] | None = None,
    prompt_session_attributes: dict[str, Any] | None = None,
    knowledge_bases_configuration: list[dict[str, Any]] | None = None,
) -> None:
    self.body = body
    self.status_code = status_code
    self.content_type = content_type
    self.session_attributes = session_attributes
    self.prompt_session_attributes = prompt_session_attributes
    self.knowledge_bases_configuration = knowledge_bases_configuration

is_json ¶

is_json() -> bool

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

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def is_json(self) -> bool:
    """
    Returns True if the response is JSON, based on the Content-Type.
    """
    return True

CORSConfig ¶

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

CORS Config

Examples:

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

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

app = APIGatewayRestResolver(cors=CORSConfig())

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

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

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

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

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

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

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

Source code in aws_lambda_powertools/event_handler/api_gateway.py

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

    self._allowed_origins = [allow_origin]

    if extra_origins:
        self._allowed_origins.extend(extra_origins)

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

build_allow_methods `staticmethod` ¶

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

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

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

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

Source code in aws_lambda_powertools/event_handler/api_gateway.py

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

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

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

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

to_dict ¶

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

Builds the configured Access-Control http headers

Source code in aws_lambda_powertools/event_handler/api_gateway.py

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

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

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

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

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

MiddlewareFrame ¶

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

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

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

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

Source code in aws_lambda_powertools/event_handler/api_gateway.py

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

ProxyEventType ¶

Bases: Enum

An enumerations of the supported proxy event types.

Response ¶

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

Bases: Generic[ResponseT]

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

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

Source code in aws_lambda_powertools/event_handler/api_gateway.py

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

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

is_json ¶

is_json() -> bool

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

Source code in aws_lambda_powertools/event_handler/api_gateway.py

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

ResponseBuilder ¶

ResponseBuilder(
    response: Response,
    serializer: Callable[[Any], str] = _JSON_DUMP_CALL,
    route: Route | None = None,
)

Bases: Generic[ResponseEventT]

Internally used Response builder

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

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def __init__(
    self,
    response: Response,
    serializer: Callable[[Any], str] = _JSON_DUMP_CALL,
    route: Route | None = None,
):
    self.response = response
    self.serializer = serializer
    self.route = route

build ¶

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

Build the full response dict to be returned by the lambda

Source code in aws_lambda_powertools/event_handler/api_gateway.py

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

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

    self._route(event, cors)

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

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

Route ¶

Route(
    method: str,
    path: str,
    rule: Pattern,
    func: Callable,
    cors: bool,
    compress: bool,
    cache_control: str | None = None,
    summary: str | None = None,
    description: str | None = None,
    responses: dict[int, OpenAPIResponse] | None = None,
    response_description: str | None = None,
    tags: list[str] | None = None,
    operation_id: str | None = None,
    include_in_schema: bool = True,
    security: list[dict[str, list[str]]] | None = None,
    openapi_extensions: dict[str, Any] | None = None,
    deprecated: bool = False,
    custom_response_validation_http_code: HTTPStatus
    | None = None,
    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`
`custom_response_validation_http_code`	Whether to have custom http status code for this route if response validation fails TYPE: `HTTPStatus \| None` DEFAULT: `None`
`middlewares`	The list of route middlewares to be called in order. TYPE: `list[Callable[..., Response]] \| None` DEFAULT: `None`

Source code in aws_lambda_powertools/event_handler/api_gateway.py

def __init__(
    self,
    method: str,
    path: str,
    rule: Pattern,
    func: Callable,
    cors: bool,
    compress: bool,
    cache_control: str | None = None,
    summary: str | None = None,
    description: str | None = None,
    responses: dict[int, OpenAPIResponse] | None = None,
    response_description: str | None = None,
    tags: list[str] | None = None,
    operation_id: str | None = None,
    include_in_schema: bool = True,
    security: list[dict[str, list[str]]] | None = None,
    openapi_extensions: dict[str, Any] | None = None,
    deprecated: bool = False,
    custom_response_validation_http_code: HTTPStatus | None = None,
    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
    custom_response_validation_http_code: int | HTTPStatus | None, optional
        Whether to have custom http status code for this route if response validation fails
    middlewares: list[Callable[..., Response]] | None
        The list of route middlewares to be called in order.
    """
    self.method = method.upper()
    self.path = path if path.strip() else "/"

    # 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

    self.custom_response_validation_http_code = custom_response_validation_http_code

Router ¶

Router()

Bases: BaseRouter

Router helper class to allow splitting ApiGatewayResolver into multiple files

Source code in aws_lambda_powertools/event_handler/api_gateway.py

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

REST

ALBResolver ¶

APIGatewayHttpResolver ¶

APIGatewayRestResolver ¶

ApiGatewayResolver ¶

configure_openapi ¶

enable_swagger ¶

get_openapi_json_schema ¶

get_openapi_schema ¶

include_router ¶

resolve ¶

Internals¶

route ¶

BaseRouter ¶

append_context ¶

clear_context ¶

delete ¶

get ¶

head ¶

patch ¶

post ¶

put ¶

use ¶

BedrockResponse ¶

is_json ¶

CORSConfig ¶

build_allow_methods staticmethod ¶

to_dict ¶

MiddlewareFrame ¶

ProxyEventType ¶

Response ¶

is_json ¶

ResponseBuilder ¶

build ¶

Route ¶

Router ¶

build_allow_methods `staticmethod` ¶