Module aws_lambda_powertools.utilities.data_classes.api_gateway_proxy_event

Classes

class APIGatewayEventAuthorizer (data: dict[str, Any], json_deserializer: Callable | None = None)

Provides a single read only access to a wrapper dict

Parameters

data : dict[str, Any]

Lambda Event Source Event payload

json_deserializer : Callable, optional

function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads

Expand source code

class APIGatewayEventAuthorizer(DictWrapper):
    @property
    def claims(self) -> dict[str, Any]:
        return self.get("claims") or {}  # key might exist but can be `null`

    @property
    def scopes(self) -> list[str]:
        return self.get("scopes") or []  # key might exist but can be `null`

    @property
    def principal_id(self) -> str:
        """The principal user identification associated with the token sent by the client and returned from an
        API Gateway Lambda authorizer (formerly known as a custom authorizer)"""
        return self.get("principalId") or ""  # key might exist but can be `null`

    @property
    def integration_latency(self) -> int | None:
        """The authorizer latency in ms."""
        return self.get("integrationLatency")

    def get_context(self) -> dict[str, Any]:
        """Retrieve the authorization context details injected by a Lambda Authorizer.

        Example
        --------

        ```python
        ctx: dict = request_context.authorizer.get_context()

        tenant_id = ctx.get("tenant_id")
        ```

        Returns:
        --------
        dict[str, Any]
            A dictionary containing Lambda authorization context details.
        """
        return self._data

Ancestors

• DictWrapper
• collections.abc.Mapping
• collections.abc.Collection
• collections.abc.Sized
• collections.abc.Iterable
• collections.abc.Container
• typing.Generic

Instance variables

prop claims : dict[str, Any]

Expand source code

@property
def claims(self) -> dict[str, Any]:
    return self.get("claims") or {}  # key might exist but can be `null`

prop integration_latency : int | None

The authorizer latency in ms.

Expand source code

@property
def integration_latency(self) -> int | None:
    """The authorizer latency in ms."""
    return self.get("integrationLatency")

prop principal_id : str

The principal user identification associated with the token sent by the client and returned from an API Gateway Lambda authorizer (formerly known as a custom authorizer)

Expand source code

@property
def principal_id(self) -> str:
    """The principal user identification associated with the token sent by the client and returned from an
    API Gateway Lambda authorizer (formerly known as a custom authorizer)"""
    return self.get("principalId") or ""  # key might exist but can be `null`

prop scopes : list[str]

Expand source code

@property
def scopes(self) -> list[str]:
    return self.get("scopes") or []  # key might exist but can be `null`

Methods

def get_context(self) ‑> dict[str, typing.Any]

Retrieve the authorization context details injected by a Lambda Authorizer.

Example

ctx: dict = request_context.authorizer.get_context()

tenant_id = ctx.get("tenant_id")

Returns:

dict[str, Any] A dictionary containing Lambda authorization context details.

Inherited members

• DictWrapper:
  • get
  • raw_event

class APIGatewayEventRequestContext (data: dict[str, Any], json_deserializer: Callable | None = None)

Provides a single read only access to a wrapper dict

Parameters

data : dict[str, Any]

Lambda Event Source Event payload

json_deserializer : Callable, optional

function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads

Expand source code

class APIGatewayEventRequestContext(BaseRequestContext):
    @property
    def connected_at(self) -> int | None:
        """The Epoch-formatted connection time. (WebSocket API)"""
        return self["requestContext"].get("connectedAt")

    @property
    def connection_id(self) -> str | None:
        """A unique ID for the connection that can be used to make a callback to the client. (WebSocket API)"""
        return self["requestContext"].get("connectionId")

    @property
    def event_type(self) -> str | None:
        """The event type: `CONNECT`, `MESSAGE`, or `DISCONNECT`. (WebSocket API)"""
        return self["requestContext"].get("eventType")

    @property
    def message_direction(self) -> str | None:
        """Message direction (WebSocket API)"""
        return self["requestContext"].get("messageDirection")

    @property
    def message_id(self) -> str | None:
        """A unique server-side ID for a message. Available only when the `eventType` is `MESSAGE`."""
        return self["requestContext"].get("messageId")

    @property
    def operation_name(self) -> str | None:
        """The name of the operation being performed"""
        return self["requestContext"].get("operationName")

    @property
    def route_key(self) -> str | None:
        """The selected route key."""
        return self["requestContext"].get("routeKey")

    @property
    def authorizer(self) -> APIGatewayEventAuthorizer:
        authz_data = self._data.get("requestContext", {}).get("authorizer", {})
        return APIGatewayEventAuthorizer(authz_data)

Ancestors

• BaseRequestContext
• DictWrapper
• collections.abc.Mapping
• collections.abc.Collection
• collections.abc.Sized
• collections.abc.Iterable
• collections.abc.Container
• typing.Generic

Instance variables

prop authorizer : APIGatewayEventAuthorizer

Expand source code

@property
def authorizer(self) -> APIGatewayEventAuthorizer:
    authz_data = self._data.get("requestContext", {}).get("authorizer", {})
    return APIGatewayEventAuthorizer(authz_data)

prop connected_at : int | None

The Epoch-formatted connection time. (WebSocket API)

Expand source code

@property
def connected_at(self) -> int | None:
    """The Epoch-formatted connection time. (WebSocket API)"""
    return self["requestContext"].get("connectedAt")

prop connection_id : str | None

A unique ID for the connection that can be used to make a callback to the client. (WebSocket API)

Expand source code

@property
def connection_id(self) -> str | None:
    """A unique ID for the connection that can be used to make a callback to the client. (WebSocket API)"""
    return self["requestContext"].get("connectionId")

prop event_type : str | None

The event type: CONNECT, MESSAGE, or DISCONNECT. (WebSocket API)

Expand source code

@property
def event_type(self) -> str | None:
    """The event type: `CONNECT`, `MESSAGE`, or `DISCONNECT`. (WebSocket API)"""
    return self["requestContext"].get("eventType")

prop message_direction : str | None

Message direction (WebSocket API)

Expand source code

@property
def message_direction(self) -> str | None:
    """Message direction (WebSocket API)"""
    return self["requestContext"].get("messageDirection")

prop message_id : str | None

A unique server-side ID for a message. Available only when the eventType is MESSAGE.

Expand source code

@property
def message_id(self) -> str | None:
    """A unique server-side ID for a message. Available only when the `eventType` is `MESSAGE`."""
    return self["requestContext"].get("messageId")

prop operation_name : str | None

The name of the operation being performed

Expand source code

@property
def operation_name(self) -> str | None:
    """The name of the operation being performed"""
    return self["requestContext"].get("operationName")

prop route_key : str | None

The selected route key.

Expand source code

@property
def route_key(self) -> str | None:
    """The selected route key."""
    return self["requestContext"].get("routeKey")

Inherited members

• BaseRequestContext:
  • account_id
  • api_id
  • domain_name
  • extended_request_id
  • get
  • http_method
  • protocol
  • raw_event
  • request_id
  • request_time
  • request_time_epoch
  • stage

class APIGatewayProxyEvent (data: dict[str, Any], json_deserializer: Callable | None = None)

AWS Lambda proxy V1

Documentation:

• https://docs.aws.amazon.com/apigateway/latest/developerguide/http-api-develop-integrations-lambda.html

Parameters

data : dict[str, Any]

Lambda Event Source Event payload

json_deserializer : Callable, optional

function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads

Expand source code

class APIGatewayProxyEvent(BaseProxyEvent):
    """AWS Lambda proxy V1

    Documentation:
    --------------
    - https://docs.aws.amazon.com/apigateway/latest/developerguide/http-api-develop-integrations-lambda.html
    """

    @property
    def version(self) -> str:
        return self["version"]

    @property
    def resource(self) -> str:
        return self["resource"]

    @property
    def multi_value_headers(self) -> dict[str, list[str]]:
        return CaseInsensitiveDict(self.get("multiValueHeaders"))

    @property
    def multi_value_query_string_parameters(self) -> dict[str, list[str]]:
        return self.get("multiValueQueryStringParameters") or {}  # key might exist but can be `null`

    @property
    def resolved_query_string_parameters(self) -> dict[str, list[str]]:
        if self.multi_value_query_string_parameters:
            return self.multi_value_query_string_parameters

        return super().resolved_query_string_parameters

    @property
    def resolved_headers_field(self) -> dict[str, Any]:
        return self.multi_value_headers or self.headers

    @property
    def request_context(self) -> APIGatewayEventRequestContext:
        return APIGatewayEventRequestContext(self._data)

    @property
    def path_parameters(self) -> dict[str, str]:
        return self.get("pathParameters") or {}

    @property
    def stage_variables(self) -> dict[str, str]:
        return self.get("stageVariables") or {}

    def header_serializer(self) -> BaseHeadersSerializer:
        return MultiValueHeadersSerializer()

Ancestors

• BaseProxyEvent
• DictWrapper
• collections.abc.Mapping
• collections.abc.Collection
• collections.abc.Sized
• collections.abc.Iterable
• collections.abc.Container
• typing.Generic

Instance variables

prop multi_value_headers : dict[str, list[str]]

Expand source code

@property
def multi_value_headers(self) -> dict[str, list[str]]:
    return CaseInsensitiveDict(self.get("multiValueHeaders"))

prop multi_value_query_string_parameters : dict[str, list[str]]

Expand source code

@property
def multi_value_query_string_parameters(self) -> dict[str, list[str]]:
    return self.get("multiValueQueryStringParameters") or {}  # key might exist but can be `null`

prop path_parameters : dict[str, str]

Expand source code

@property
def path_parameters(self) -> dict[str, str]:
    return self.get("pathParameters") or {}

prop request_context : APIGatewayEventRequestContext

Expand source code

@property
def request_context(self) -> APIGatewayEventRequestContext:
    return APIGatewayEventRequestContext(self._data)

prop resource : str

Expand source code

@property
def resource(self) -> str:
    return self["resource"]

prop stage_variables : dict[str, str]

Expand source code

@property
def stage_variables(self) -> dict[str, str]:
    return self.get("stageVariables") or {}

prop version : str

Expand source code

@property
def version(self) -> str:
    return self["version"]

Methods

def header_serializer(self) ‑> BaseHeadersSerializer

Inherited members

• BaseProxyEvent:
  • body
  • decoded_body
  • get
  • get_header_value
  • get_multi_value_query_string_values
  • get_query_string_value
  • http_method
  • json_body
  • raw_event
  • resolved_headers_field
  • resolved_query_string_parameters

class APIGatewayProxyEventV2 (data: dict[str, Any], json_deserializer: Callable | None = None)

AWS Lambda proxy V2 event

Notes:

Format 2.0 doesn't have multiValueHeaders or multiValueQueryStringParameters fields. Duplicate headers are combined with commas and included in the headers field. Duplicate query strings are combined with commas and included in the queryStringParameters field.

Format 2.0 includes a new cookies field. All cookie headers in the request are combined with commas and added to the cookies field. In the response to the client, each cookie becomes a set-cookie header.

Documentation:

• https://docs.aws.amazon.com/apigateway/latest/developerguide/http-api-develop-integrations-lambda.html

Parameters

data : dict[str, Any]

Lambda Event Source Event payload

json_deserializer : Callable, optional

function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads

Expand source code

class APIGatewayProxyEventV2(BaseProxyEvent):
    """AWS Lambda proxy V2 event

    Notes:
    -----
    Format 2.0 doesn't have multiValueHeaders or multiValueQueryStringParameters fields. Duplicate headers
    are combined with commas and included in the headers field. Duplicate query strings are combined with
    commas and included in the queryStringParameters field.

    Format 2.0 includes a new cookies field. All cookie headers in the request are combined with commas and
    added to the cookies field. In the response to the client, each cookie becomes a set-cookie header.

    Documentation:
    --------------
    - https://docs.aws.amazon.com/apigateway/latest/developerguide/http-api-develop-integrations-lambda.html
    """

    @property
    def version(self) -> str:
        return self["version"]

    @property
    def route_key(self) -> str:
        return self["routeKey"]

    @property
    def raw_path(self) -> str:
        return self["rawPath"]

    @property
    def raw_query_string(self) -> str:
        return self["rawQueryString"]

    @property
    def cookies(self) -> list[str]:
        return self.get("cookies") or []

    @property
    def request_context(self) -> RequestContextV2:
        return RequestContextV2(self._data)

    @property
    def path_parameters(self) -> dict[str, str]:
        return self.get("pathParameters") or {}

    @property
    def stage_variables(self) -> dict[str, str]:
        return self.get("stageVariables") or {}

    @property
    def path(self) -> str:
        stage = self.request_context.stage
        if stage != "$default":
            return self.raw_path[len("/" + stage) :]
        return self.raw_path

    @property
    def http_method(self) -> str:
        """The HTTP method used. Valid values include: DELETE, GET, HEAD, OPTIONS, PATCH, POST, and PUT."""
        return self.request_context.http.method

    def header_serializer(self):
        return HttpApiHeadersSerializer()

    @cached_property
    def resolved_headers_field(self) -> dict[str, Any]:
        return CaseInsensitiveDict((k, v.split(",") if "," in v else v) for k, v in self.headers.items())

Ancestors

• BaseProxyEvent
• DictWrapper
• collections.abc.Mapping
• collections.abc.Collection
• collections.abc.Sized
• collections.abc.Iterable
• collections.abc.Container
• typing.Generic

Subclasses

• LambdaFunctionUrlEvent

Instance variables

prop cookies : list[str]

Expand source code

@property
def cookies(self) -> list[str]:
    return self.get("cookies") or []

prop path : str

Expand source code

@property
def path(self) -> str:
    stage = self.request_context.stage
    if stage != "$default":
        return self.raw_path[len("/" + stage) :]
    return self.raw_path

prop path_parameters : dict[str, str]

Expand source code

@property
def path_parameters(self) -> dict[str, str]:
    return self.get("pathParameters") or {}

prop raw_path : str

Expand source code

@property
def raw_path(self) -> str:
    return self["rawPath"]

prop raw_query_string : str

Expand source code

@property
def raw_query_string(self) -> str:
    return self["rawQueryString"]

prop request_context : RequestContextV2

Expand source code

@property
def request_context(self) -> RequestContextV2:
    return RequestContextV2(self._data)

prop route_key : str

Expand source code

@property
def route_key(self) -> str:
    return self["routeKey"]

prop stage_variables : dict[str, str]

Expand source code

@property
def stage_variables(self) -> dict[str, str]:
    return self.get("stageVariables") or {}

prop version : str

Expand source code

@property
def version(self) -> str:
    return self["version"]

Methods

def header_serializer(self)

Inherited members

• BaseProxyEvent:
  • body
  • decoded_body
  • get
  • get_header_value
  • get_multi_value_query_string_values
  • get_query_string_value
  • http_method
  • json_body
  • raw_event
  • resolved_headers_field
  • resolved_query_string_parameters

class RequestContextV2 (data: dict[str, Any], json_deserializer: Callable | None = None)

Provides a single read only access to a wrapper dict

Parameters

data : dict[str, Any]

Lambda Event Source Event payload

json_deserializer : Callable, optional

function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads

Expand source code

class RequestContextV2(BaseRequestContextV2):
    @property
    def authorizer(self) -> RequestContextV2Authorizer:
        ctx = self.get("requestContext") or {}  # key might exist but can be `null`
        return RequestContextV2Authorizer(ctx.get("authorizer", {}))

Ancestors

• BaseRequestContextV2
• DictWrapper
• collections.abc.Mapping
• collections.abc.Collection
• collections.abc.Sized
• collections.abc.Iterable
• collections.abc.Container
• typing.Generic

Instance variables

prop authorizer : RequestContextV2Authorizer

Expand source code

@property
def authorizer(self) -> RequestContextV2Authorizer:
    ctx = self.get("requestContext") or {}  # key might exist but can be `null`
    return RequestContextV2Authorizer(ctx.get("authorizer", {}))

Inherited members

• BaseRequestContextV2:
  • account_id
  • api_id
  • authentication
  • domain_name
  • get
  • raw_event
  • request_id
  • route_key
  • stage
  • time
  • time_epoch

class RequestContextV2Authorizer (data: dict[str, Any], json_deserializer: Callable | None = None)

Provides a single read only access to a wrapper dict

Parameters

data : dict[str, Any]

Lambda Event Source Event payload

json_deserializer : Callable, optional

function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads

Expand source code

class RequestContextV2Authorizer(DictWrapper):
    @property
    def jwt_claim(self) -> dict[str, Any]:
        jwt = self.get("jwt") or {}  # not available in FunctionURL; key might exist but can be `null`
        return jwt.get("claims") or {}  # key might exist but can be `null`

    @property
    def jwt_scopes(self) -> list[str]:
        jwt = self.get("jwt") or {}  # not available in FunctionURL; key might exist but can be `null`
        return jwt.get("scopes", [])

    @property
    def get_lambda(self) -> dict[str, Any]:
        """Lambda authorization context details"""
        return self.get("lambda") or {}  # key might exist but can be `null`

    def get_context(self) -> dict[str, Any]:
        """Retrieve the authorization context details injected by a Lambda Authorizer.

        Example
        --------

        ```python
        ctx: dict = request_context.authorizer.get_context()

        tenant_id = ctx.get("tenant_id")
        ```

        Returns:
        --------
        dict[str, Any]
            A dictionary containing Lambda authorization context details.
        """
        return self.get_lambda

    @property
    def iam(self) -> RequestContextV2AuthorizerIam:
        """IAM authorization details used for making the request."""
        iam = self.get("iam") or {}  # key might exist but can be `null`
        return RequestContextV2AuthorizerIam(iam)

Ancestors

• DictWrapper
• collections.abc.Mapping
• collections.abc.Collection
• collections.abc.Sized
• collections.abc.Iterable
• collections.abc.Container
• typing.Generic

Instance variables

prop get_lambda : dict[str, Any]

Lambda authorization context details

Expand source code

@property
def get_lambda(self) -> dict[str, Any]:
    """Lambda authorization context details"""
    return self.get("lambda") or {}  # key might exist but can be `null`

prop iam : RequestContextV2AuthorizerIam

IAM authorization details used for making the request.

Expand source code

@property
def iam(self) -> RequestContextV2AuthorizerIam:
    """IAM authorization details used for making the request."""
    iam = self.get("iam") or {}  # key might exist but can be `null`
    return RequestContextV2AuthorizerIam(iam)

prop jwt_claim : dict[str, Any]

Expand source code

@property
def jwt_claim(self) -> dict[str, Any]:
    jwt = self.get("jwt") or {}  # not available in FunctionURL; key might exist but can be `null`
    return jwt.get("claims") or {}  # key might exist but can be `null`

prop jwt_scopes : list[str]

Expand source code

@property
def jwt_scopes(self) -> list[str]:
    jwt = self.get("jwt") or {}  # not available in FunctionURL; key might exist but can be `null`
    return jwt.get("scopes", [])

Methods

def get_context(self) ‑> dict[str, typing.Any]

Retrieve the authorization context details injected by a Lambda Authorizer.

Example

ctx: dict = request_context.authorizer.get_context()

tenant_id = ctx.get("tenant_id")

Returns:

dict[str, Any] A dictionary containing Lambda authorization context details.

Inherited members

• DictWrapper:
  • get
  • raw_event

class RequestContextV2AuthorizerIam (data: dict[str, Any], json_deserializer: Callable | None = None)

Provides a single read only access to a wrapper dict

Parameters

data : dict[str, Any]

Lambda Event Source Event payload

json_deserializer : Callable, optional

function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads

Expand source code

class RequestContextV2AuthorizerIam(DictWrapper):
    @property
    def access_key(self) -> str:
        """The IAM user access key associated with the request."""
        return self.get("accessKey") or ""  # key might exist but can be `null`

    @property
    def account_id(self) -> str:
        """The AWS account ID associated with the request."""
        return self.get("accountId") or ""  # key might exist but can be `null`

    @property
    def caller_id(self) -> str:
        """The principal identifier of the caller making the request."""
        return self.get("callerId") or ""  # key might exist but can be `null`

    def _cognito_identity(self) -> dict:
        return self.get("cognitoIdentity") or {}  # not available in FunctionURL; key might exist but can be `null`

    @property
    def cognito_amr(self) -> list[str]:
        """This represents how the user was authenticated.
        AMR stands for  Authentication Methods References as per the openid spec"""
        return self._cognito_identity().get("amr", [])

    @property
    def cognito_identity_id(self) -> str:
        """The Amazon Cognito identity ID of the caller making the request.
        Available only if the request was signed with Amazon Cognito credentials."""
        return self._cognito_identity().get("identityId", "")

    @property
    def cognito_identity_pool_id(self) -> str:
        """The Amazon Cognito identity pool ID of the caller making the request.
        Available only if the request was signed with Amazon Cognito credentials."""
        return self._cognito_identity().get("identityPoolId") or ""  # key might exist but can be `null`

    @property
    def principal_org_id(self) -> str:
        """The AWS organization ID."""
        return self.get("principalOrgId") or ""  # key might exist but can be `null`

    @property
    def user_arn(self) -> str:
        """The Amazon Resource Name (ARN) of the effective user identified after authentication."""
        return self.get("userArn") or ""  # key might exist but can be `null`

    @property
    def user_id(self) -> str:
        """The IAM user ID of the effective user identified after authentication."""
        return self.get("userId") or ""  # key might exist but can be `null`

Ancestors

• DictWrapper
• collections.abc.Mapping
• collections.abc.Collection
• collections.abc.Sized
• collections.abc.Iterable
• collections.abc.Container
• typing.Generic

Instance variables

prop access_key : str

The IAM user access key associated with the request.

Expand source code

@property
def access_key(self) -> str:
    """The IAM user access key associated with the request."""
    return self.get("accessKey") or ""  # key might exist but can be `null`

prop account_id : str

The AWS account ID associated with the request.

Expand source code

@property
def account_id(self) -> str:
    """The AWS account ID associated with the request."""
    return self.get("accountId") or ""  # key might exist but can be `null`

prop caller_id : str

The principal identifier of the caller making the request.

Expand source code

@property
def caller_id(self) -> str:
    """The principal identifier of the caller making the request."""
    return self.get("callerId") or ""  # key might exist but can be `null`

prop cognito_amr : list[str]

This represents how the user was authenticated. AMR stands for Authentication Methods References as per the openid spec

Expand source code

@property
def cognito_amr(self) -> list[str]:
    """This represents how the user was authenticated.
    AMR stands for  Authentication Methods References as per the openid spec"""
    return self._cognito_identity().get("amr", [])

prop cognito_identity_id : str

The Amazon Cognito identity ID of the caller making the request. Available only if the request was signed with Amazon Cognito credentials.

Expand source code

@property
def cognito_identity_id(self) -> str:
    """The Amazon Cognito identity ID of the caller making the request.
    Available only if the request was signed with Amazon Cognito credentials."""
    return self._cognito_identity().get("identityId", "")

prop cognito_identity_pool_id : str

The Amazon Cognito identity pool ID of the caller making the request. Available only if the request was signed with Amazon Cognito credentials.

Expand source code

@property
def cognito_identity_pool_id(self) -> str:
    """The Amazon Cognito identity pool ID of the caller making the request.
    Available only if the request was signed with Amazon Cognito credentials."""
    return self._cognito_identity().get("identityPoolId") or ""  # key might exist but can be `null`

prop principal_org_id : str

The AWS organization ID.

Expand source code

@property
def principal_org_id(self) -> str:
    """The AWS organization ID."""
    return self.get("principalOrgId") or ""  # key might exist but can be `null`

prop user_arn : str

The Amazon Resource Name (ARN) of the effective user identified after authentication.

Expand source code

@property
def user_arn(self) -> str:
    """The Amazon Resource Name (ARN) of the effective user identified after authentication."""
    return self.get("userArn") or ""  # key might exist but can be `null`

prop user_id : str

The IAM user ID of the effective user identified after authentication.

Expand source code

@property
def user_id(self) -> str:
    """The IAM user ID of the effective user identified after authentication."""
    return self.get("userId") or ""  # key might exist but can be `null`

Inherited members

• DictWrapper:
  • get
  • raw_event