Module aws_lambda_powertools.utilities.data_classes.common

Classes

class APIGatewayEventIdentity (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 APIGatewayEventIdentity(DictWrapper):
    @property
    def access_key(self) -> str | None:
        return self["requestContext"]["identity"].get("accessKey")

    @property
    def account_id(self) -> str | None:
        """The AWS account ID associated with the request."""
        return self["requestContext"]["identity"].get("accountId")

    @property
    def api_key(self) -> str | None:
        """For API methods that require an API key, this variable is the API key associated with the method request.
        For methods that don't require an API key, this variable is null."""
        return self["requestContext"]["identity"].get("apiKey")

    @property
    def api_key_id(self) -> str | None:
        """The API key ID associated with an API request that requires an API key."""
        return self["requestContext"]["identity"].get("apiKeyId")

    @property
    def caller(self) -> str | None:
        """The principal identifier of the caller making the request."""
        return self["requestContext"]["identity"].get("caller")

    @property
    def cognito_authentication_provider(self) -> str | None:
        """A comma-separated list of the Amazon Cognito authentication providers used by the caller
        making the request. Available only if the request was signed with Amazon Cognito credentials."""
        return self["requestContext"]["identity"].get("cognitoAuthenticationProvider")

    @property
    def cognito_authentication_type(self) -> str | None:
        """The Amazon Cognito authentication type of the caller making the request.
        Available only if the request was signed with Amazon Cognito credentials."""
        return self["requestContext"]["identity"].get("cognitoAuthenticationType")

    @property
    def cognito_identity_id(self) -> str | None:
        """The Amazon Cognito identity ID of the caller making the request.
        Available only if the request was signed with Amazon Cognito credentials."""
        return self["requestContext"]["identity"].get("cognitoIdentityId")

    @property
    def cognito_identity_pool_id(self) -> str | None:
        """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["requestContext"]["identity"].get("cognitoIdentityPoolId")

    @property
    def principal_org_id(self) -> str | None:
        """The AWS organization ID."""
        return self["requestContext"]["identity"].get("principalOrgId")

    @property
    def source_ip(self) -> str:
        """The source IP address of the TCP connection making the request to API Gateway."""
        return self["requestContext"]["identity"]["sourceIp"]

    @property
    def user(self) -> str | None:
        """The principal identifier of the user making the request."""
        return self["requestContext"]["identity"].get("user")

    @property
    def user_agent(self) -> str | None:
        """The User Agent of the API caller."""
        return self["requestContext"]["identity"].get("userAgent")

    @property
    def user_arn(self) -> str | None:
        """The Amazon Resource Name (ARN) of the effective user identified after authentication."""
        return self["requestContext"]["identity"].get("userArn")

    @property
    def client_cert(self) -> RequestContextClientCert | None:
        client_cert = self["requestContext"]["identity"].get("clientCert")
        return None if client_cert is None else RequestContextClientCert(client_cert)

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 | None

Expand source code

@property
def access_key(self) -> str | None:
    return self["requestContext"]["identity"].get("accessKey")

prop account_id : str | None

The AWS account ID associated with the request.

Expand source code

@property
def account_id(self) -> str | None:
    """The AWS account ID associated with the request."""
    return self["requestContext"]["identity"].get("accountId")

prop api_key : str | None

For API methods that require an API key, this variable is the API key associated with the method request. For methods that don't require an API key, this variable is null.

Expand source code

@property
def api_key(self) -> str | None:
    """For API methods that require an API key, this variable is the API key associated with the method request.
    For methods that don't require an API key, this variable is null."""
    return self["requestContext"]["identity"].get("apiKey")

prop api_key_id : str | None

The API key ID associated with an API request that requires an API key.

Expand source code

@property
def api_key_id(self) -> str | None:
    """The API key ID associated with an API request that requires an API key."""
    return self["requestContext"]["identity"].get("apiKeyId")

prop caller : str | None

The principal identifier of the caller making the request.

Expand source code

@property
def caller(self) -> str | None:
    """The principal identifier of the caller making the request."""
    return self["requestContext"]["identity"].get("caller")

prop client_cert : RequestContextClientCert | None

Expand source code

@property
def client_cert(self) -> RequestContextClientCert | None:
    client_cert = self["requestContext"]["identity"].get("clientCert")
    return None if client_cert is None else RequestContextClientCert(client_cert)

prop cognito_authentication_provider : str | None

A comma-separated list of the Amazon Cognito authentication providers used by the caller making the request. Available only if the request was signed with Amazon Cognito credentials.

Expand source code

@property
def cognito_authentication_provider(self) -> str | None:
    """A comma-separated list of the Amazon Cognito authentication providers used by the caller
    making the request. Available only if the request was signed with Amazon Cognito credentials."""
    return self["requestContext"]["identity"].get("cognitoAuthenticationProvider")

prop cognito_authentication_type : str | None

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

Expand source code

@property
def cognito_authentication_type(self) -> str | None:
    """The Amazon Cognito authentication type of the caller making the request.
    Available only if the request was signed with Amazon Cognito credentials."""
    return self["requestContext"]["identity"].get("cognitoAuthenticationType")

prop cognito_identity_id : str | None

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 | None:
    """The Amazon Cognito identity ID of the caller making the request.
    Available only if the request was signed with Amazon Cognito credentials."""
    return self["requestContext"]["identity"].get("cognitoIdentityId")

prop cognito_identity_pool_id : str | None

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 | None:
    """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["requestContext"]["identity"].get("cognitoIdentityPoolId")

prop principal_org_id : str | None

The AWS organization ID.

Expand source code

@property
def principal_org_id(self) -> str | None:
    """The AWS organization ID."""
    return self["requestContext"]["identity"].get("principalOrgId")

prop source_ip : str

The source IP address of the TCP connection making the request to API Gateway.

Expand source code

@property
def source_ip(self) -> str:
    """The source IP address of the TCP connection making the request to API Gateway."""
    return self["requestContext"]["identity"]["sourceIp"]

prop user : str | None

The principal identifier of the user making the request.

Expand source code

@property
def user(self) -> str | None:
    """The principal identifier of the user making the request."""
    return self["requestContext"]["identity"].get("user")

prop user_agent : str | None

The User Agent of the API caller.

Expand source code

@property
def user_agent(self) -> str | None:
    """The User Agent of the API caller."""
    return self["requestContext"]["identity"].get("userAgent")

prop user_arn : str | None

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

Expand source code

@property
def user_arn(self) -> str | None:
    """The Amazon Resource Name (ARN) of the effective user identified after authentication."""
    return self["requestContext"]["identity"].get("userArn")

Inherited members

• DictWrapper:
  • get
  • raw_event

class BaseProxyEvent (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 BaseProxyEvent(DictWrapper):
    @property
    def headers(self) -> dict[str, str]:
        return CaseInsensitiveDict(self.get("headers"))

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

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

    @cached_property
    def resolved_query_string_parameters(self) -> dict[str, list[str]]:
        """
        This property determines the appropriate query string parameter to be used
        as a trusted source for validating OpenAPI.

        This is necessary because different resolvers use different formats to encode
        multi query string parameters.
        """
        return {k: v.split(",") for k, v in self.query_string_parameters.items()}

    @property
    def resolved_headers_field(self) -> dict[str, str]:
        """
        This property determines the appropriate header to be used
        as a trusted source for validating OpenAPI.

        This is necessary because different resolvers use different formats to encode
        headers parameters.

        Headers are case-insensitive according to RFC 7540 (HTTP/2), so we lower the header name
        This ensures that customers can access headers with any casing, as per the RFC guidelines.
        Reference: https://www.rfc-editor.org/rfc/rfc7540#section-8.1.2
        """
        return self.headers

    @property
    def is_base64_encoded(self) -> bool | None:
        return self.get("isBase64Encoded")

    @property
    def body(self) -> str | None:
        """Submitted body of the request as a string"""
        return self.get("body")

    @cached_property
    def json_body(self) -> Any:
        """Parses the submitted body as json"""
        if self.decoded_body:
            return self._json_deserializer(self.decoded_body)

        return None

    @cached_property
    def decoded_body(self) -> str | None:
        """Decode the body from base64 if encoded, otherwise return it as is."""
        body: str | None = self.body
        if self.is_base64_encoded and body:
            return base64.b64decode(body.encode()).decode()
        return body

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

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

    @overload
    def get_query_string_value(self, name: str, default_value: str) -> str: ...

    @overload
    def get_query_string_value(self, name: str, default_value: str | None = None) -> str | None: ...

    def get_query_string_value(self, name: str, default_value: str | None = None) -> str | None:
        """Get query string value by name
        Parameters
        ----------
        name: str
            Query string parameter name
        default_value: str, optional
            Default value if no value was found by name
        Returns
        -------
        str, optional
            Query string parameter value
        """
        return get_query_string_value(
            query_string_parameters=self.query_string_parameters,
            name=name,
            default_value=default_value,
        )

    def get_multi_value_query_string_values(
        self,
        name: str,
        default_values: list[str] | None = None,
    ) -> list[str]:
        """Get multi-value query string parameter values by name
        Parameters
        ----------
        name: str
            Multi-Value query string parameter name
        default_values: List[str], optional
            Default values is no values are found by name
        Returns
        -------
        List[str], optional
            List of query string values
        """
        return get_multi_value_query_string_values(
            multi_value_query_string_parameters=self.multi_value_query_string_parameters,
            name=name,
            default_values=default_values,
        )

    @overload
    def get_header_value(
        self,
        name: str,
        default_value: str,
        case_sensitive: bool = False,
    ) -> str: ...

    @overload
    def get_header_value(
        self,
        name: str,
        default_value: str | None = None,
        case_sensitive: bool = False,
    ) -> str | None: ...

    @deprecated(
        "`get_header_value` function is deprecated; Access headers directly using event.headers.get('HeaderName')",
        category=None,
    )
    def get_header_value(
        self,
        name: str,
        default_value: str | None = None,
        case_sensitive: bool = False,
    ) -> str | None:
        """Get header value by name
        Parameters
        ----------
        name: str
            Header name
        default_value: str, optional
            Default value if no value was found by name
        case_sensitive: bool
            Whether to use a case-sensitive look up. By default we make a case-insensitive lookup.
        Returns
        -------
        str, optional
            Header value
        """
        warnings.warn(
            "The `get_header_value` function is deprecated in V3 and the `case_sensitive` parameter "
            "no longer has any effect. This function will be removed in the next major version. "
            "Instead, access headers directly using event.headers.get('HeaderName'), which is case insensitive.",
            category=PowertoolsDeprecationWarning,
            stacklevel=2,
        )
        return get_header_value(
            headers=self.headers,
            name=name,
            default_value=default_value,
            case_sensitive=case_sensitive,
        )

    def header_serializer(self) -> BaseHeadersSerializer:
        raise NotImplementedError()

Ancestors

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

Subclasses

• ALBEvent
• APIGatewayProxyEvent
• APIGatewayProxyEventV2
• BedrockAgentEvent
• VPCLatticeEventBase

Instance variables

prop body : str | None

Submitted body of the request as a string

Expand source code

@property
def body(self) -> str | None:
    """Submitted body of the request as a string"""
    return self.get("body")

var decoded_body

Decode the body from base64 if encoded, otherwise return it as is.

Expand source code

def __get__(self, instance, owner=None):
    if instance is None:
        return self
    if self.attrname is None:
        raise TypeError(
            "Cannot use cached_property instance without calling __set_name__ on it.")
    try:
        cache = instance.__dict__
    except AttributeError:  # not all objects have __dict__ (e.g. class defines slots)
        msg = (
            f"No '__dict__' attribute on {type(instance).__name__!r} "
            f"instance to cache {self.attrname!r} property."
        )
        raise TypeError(msg) from None
    val = cache.get(self.attrname, _NOT_FOUND)
    if val is _NOT_FOUND:
        val = self.func(instance)
        try:
            cache[self.attrname] = val
        except TypeError:
            msg = (
                f"The '__dict__' attribute on {type(instance).__name__!r} instance "
                f"does not support item assignment for caching {self.attrname!r} property."
            )
            raise TypeError(msg) from None
    return val

prop headers : dict[str, str]

Expand source code

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

prop http_method : str

The HTTP method used. Valid values include: DELETE, GET, HEAD, OPTIONS, PATCH, POST, and PUT.

Expand source code

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

prop is_base64_encoded : bool | None

Expand source code

@property
def is_base64_encoded(self) -> bool | None:
    return self.get("isBase64Encoded")

var json_body

Parses the submitted body as json

Expand source code

def __get__(self, instance, owner=None):
    if instance is None:
        return self
    if self.attrname is None:
        raise TypeError(
            "Cannot use cached_property instance without calling __set_name__ on it.")
    try:
        cache = instance.__dict__
    except AttributeError:  # not all objects have __dict__ (e.g. class defines slots)
        msg = (
            f"No '__dict__' attribute on {type(instance).__name__!r} "
            f"instance to cache {self.attrname!r} property."
        )
        raise TypeError(msg) from None
    val = cache.get(self.attrname, _NOT_FOUND)
    if val is _NOT_FOUND:
        val = self.func(instance)
        try:
            cache[self.attrname] = val
        except TypeError:
            msg = (
                f"The '__dict__' attribute on {type(instance).__name__!r} instance "
                f"does not support item assignment for caching {self.attrname!r} property."
            )
            raise TypeError(msg) from None
    return val

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 {}

prop path : str

Expand source code

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

prop query_string_parameters : dict[str, str]

Expand source code

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

prop resolved_headers_field : dict[str, str]

This property determines the appropriate header to be used as a trusted source for validating OpenAPI.

This is necessary because different resolvers use different formats to encode headers parameters.

Headers are case-insensitive according to RFC 7540 (HTTP/2), so we lower the header name This ensures that customers can access headers with any casing, as per the RFC guidelines. Reference: https://www.rfc-editor.org/rfc/rfc7540#section-8.1.2

Expand source code

@property
def resolved_headers_field(self) -> dict[str, str]:
    """
    This property determines the appropriate header to be used
    as a trusted source for validating OpenAPI.

    This is necessary because different resolvers use different formats to encode
    headers parameters.

    Headers are case-insensitive according to RFC 7540 (HTTP/2), so we lower the header name
    This ensures that customers can access headers with any casing, as per the RFC guidelines.
    Reference: https://www.rfc-editor.org/rfc/rfc7540#section-8.1.2
    """
    return self.headers

var resolved_query_string_parameters

This property determines the appropriate query string parameter to be used as a trusted source for validating OpenAPI.

This is necessary because different resolvers use different formats to encode multi query string parameters.

Expand source code

def __get__(self, instance, owner=None):
    if instance is None:
        return self
    if self.attrname is None:
        raise TypeError(
            "Cannot use cached_property instance without calling __set_name__ on it.")
    try:
        cache = instance.__dict__
    except AttributeError:  # not all objects have __dict__ (e.g. class defines slots)
        msg = (
            f"No '__dict__' attribute on {type(instance).__name__!r} "
            f"instance to cache {self.attrname!r} property."
        )
        raise TypeError(msg) from None
    val = cache.get(self.attrname, _NOT_FOUND)
    if val is _NOT_FOUND:
        val = self.func(instance)
        try:
            cache[self.attrname] = val
        except TypeError:
            msg = (
                f"The '__dict__' attribute on {type(instance).__name__!r} instance "
                f"does not support item assignment for caching {self.attrname!r} property."
            )
            raise TypeError(msg) from None
    return val

Methods

def get_header_value(self, name: str, default_value: str | None = None, case_sensitive: bool = False) ‑> str | None

Get header value by name Parameters

---

name : str

Header name

default_value : str, optional

Default value if no value was found by name

case_sensitive : bool

Whether to use a case-sensitive look up. By default we make a case-insensitive lookup.

Returns

str, optional

Header value

def get_multi_value_query_string_values(self, name: str, default_values: list[str] | None = None) ‑> list[str]

Get multi-value query string parameter values by name Parameters

---

name : str

Multi-Value query string parameter name

default_values : List[str], optional

Default values is no values are found by name

Returns

List[str], optional

List of query string values

def get_query_string_value(self, name: str, default_value: str | None = None) ‑> str | None

Get query string value by name Parameters

---

name : str

Query string parameter name

default_value : str, optional

Default value if no value was found by name

Returns

str, optional

Query string parameter value

def header_serializer(self)

Inherited members

• DictWrapper:
  • get
  • raw_event

class BaseRequestContext (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 BaseRequestContext(DictWrapper):
    @property
    def account_id(self) -> str:
        """The AWS account ID associated with the request."""
        return self["requestContext"]["accountId"]

    @property
    def api_id(self) -> str:
        """The identifier API Gateway assigns to your API."""
        return self["requestContext"]["apiId"]

    @property
    def domain_name(self) -> str | None:
        """A domain name"""
        return self["requestContext"].get("domainName")

    @property
    def domain_prefix(self) -> str | None:
        return self["requestContext"].get("domainPrefix")

    @property
    def extended_request_id(self) -> str | None:
        """An automatically generated ID for the API call, which contains more useful information
        for debugging/troubleshooting."""
        return self["requestContext"].get("extendedRequestId")

    @property
    def protocol(self) -> str:
        """The request protocol, for example, HTTP/1.1."""
        return self["requestContext"]["protocol"]

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

    @property
    def identity(self) -> APIGatewayEventIdentity:
        return APIGatewayEventIdentity(self._data)

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

    @property
    def stage(self) -> str:
        """The deployment stage of the API request"""
        return self["requestContext"]["stage"]

    @property
    def request_id(self) -> str:
        """The ID that API Gateway assigns to the API request."""
        return self["requestContext"]["requestId"]

    @property
    def request_time(self) -> str | None:
        """The CLF-formatted request time (dd/MMM/yyyy:HH:mm:ss +-hhmm)"""
        return self["requestContext"].get("requestTime")

    @property
    def request_time_epoch(self) -> int:
        """The Epoch-formatted request time."""
        return self["requestContext"]["requestTimeEpoch"]

    @property
    def resource_id(self) -> str:
        return self["requestContext"]["resourceId"]

    @property
    def resource_path(self) -> str:
        return self["requestContext"]["resourcePath"]

Ancestors

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

Subclasses

• APIGatewayEventRequestContext

Instance variables

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["requestContext"]["accountId"]

prop api_id : str

The identifier API Gateway assigns to your API.

Expand source code

@property
def api_id(self) -> str:
    """The identifier API Gateway assigns to your API."""
    return self["requestContext"]["apiId"]

prop domain_name : str | None

A domain name

Expand source code

@property
def domain_name(self) -> str | None:
    """A domain name"""
    return self["requestContext"].get("domainName")

prop domain_prefix : str | None

Expand source code

@property
def domain_prefix(self) -> str | None:
    return self["requestContext"].get("domainPrefix")

prop extended_request_id : str | None

An automatically generated ID for the API call, which contains more useful information for debugging/troubleshooting.

Expand source code

@property
def extended_request_id(self) -> str | None:
    """An automatically generated ID for the API call, which contains more useful information
    for debugging/troubleshooting."""
    return self["requestContext"].get("extendedRequestId")

prop http_method : str

The HTTP method used. Valid values include: DELETE, GET, HEAD, OPTIONS, PATCH, POST, and PUT.

Expand source code

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

prop identity : APIGatewayEventIdentity

Expand source code

@property
def identity(self) -> APIGatewayEventIdentity:
    return APIGatewayEventIdentity(self._data)

prop path : str

Expand source code

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

prop protocol : str

The request protocol, for example, HTTP/1.1.

Expand source code

@property
def protocol(self) -> str:
    """The request protocol, for example, HTTP/1.1."""
    return self["requestContext"]["protocol"]

prop request_id : str

The ID that API Gateway assigns to the API request.

Expand source code

@property
def request_id(self) -> str:
    """The ID that API Gateway assigns to the API request."""
    return self["requestContext"]["requestId"]

prop request_time : str | None

The CLF-formatted request time (dd/MMM/yyyy:HH:mm:ss +-hhmm)

Expand source code

@property
def request_time(self) -> str | None:
    """The CLF-formatted request time (dd/MMM/yyyy:HH:mm:ss +-hhmm)"""
    return self["requestContext"].get("requestTime")

prop request_time_epoch : int

The Epoch-formatted request time.

Expand source code

@property
def request_time_epoch(self) -> int:
    """The Epoch-formatted request time."""
    return self["requestContext"]["requestTimeEpoch"]

prop resource_id : str

Expand source code

@property
def resource_id(self) -> str:
    return self["requestContext"]["resourceId"]

prop resource_path : str

Expand source code

@property
def resource_path(self) -> str:
    return self["requestContext"]["resourcePath"]

prop stage : str

The deployment stage of the API request

Expand source code

@property
def stage(self) -> str:
    """The deployment stage of the API request"""
    return self["requestContext"]["stage"]

Inherited members

• DictWrapper:
  • get
  • raw_event

class BaseRequestContextV2 (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 BaseRequestContextV2(DictWrapper):
    @property
    def account_id(self) -> str:
        """The AWS account ID associated with the request."""
        return self["requestContext"]["accountId"]

    @property
    def api_id(self) -> str:
        """The identifier API Gateway assigns to your API."""
        return self["requestContext"]["apiId"]

    @property
    def domain_name(self) -> str:
        """A domain name"""
        return self["requestContext"]["domainName"]

    @property
    def domain_prefix(self) -> str:
        return self["requestContext"]["domainPrefix"]

    @property
    def http(self) -> RequestContextV2Http:
        return RequestContextV2Http(self._data)

    @property
    def request_id(self) -> str:
        """The ID that API Gateway assigns to the API request."""
        return self["requestContext"]["requestId"]

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

    @property
    def stage(self) -> str:
        """The deployment stage of the API request"""
        return self["requestContext"]["stage"]

    @property
    def time(self) -> str:
        """The CLF-formatted request time (dd/MMM/yyyy:HH:mm:ss +-hhmm)."""
        return self["requestContext"]["time"]

    @property
    def time_epoch(self) -> int:
        """The Epoch-formatted request time."""
        return self["requestContext"]["timeEpoch"]

    @property
    def authentication(self) -> RequestContextClientCert | None:
        """Optional when using mutual TLS authentication"""
        # FunctionURL might have NONE as AuthZ
        authentication = self["requestContext"].get("authentication") or {}
        client_cert = authentication.get("clientCert")
        return None if client_cert is None else RequestContextClientCert(client_cert)

Ancestors

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

Subclasses

• RequestContextV2

Instance variables

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["requestContext"]["accountId"]

prop api_id : str

The identifier API Gateway assigns to your API.

Expand source code

@property
def api_id(self) -> str:
    """The identifier API Gateway assigns to your API."""
    return self["requestContext"]["apiId"]

prop authentication : RequestContextClientCert | None

Optional when using mutual TLS authentication

Expand source code

@property
def authentication(self) -> RequestContextClientCert | None:
    """Optional when using mutual TLS authentication"""
    # FunctionURL might have NONE as AuthZ
    authentication = self["requestContext"].get("authentication") or {}
    client_cert = authentication.get("clientCert")
    return None if client_cert is None else RequestContextClientCert(client_cert)

prop domain_name : str

A domain name

Expand source code

@property
def domain_name(self) -> str:
    """A domain name"""
    return self["requestContext"]["domainName"]

prop domain_prefix : str

Expand source code

@property
def domain_prefix(self) -> str:
    return self["requestContext"]["domainPrefix"]

prop http : RequestContextV2Http

Expand source code

@property
def http(self) -> RequestContextV2Http:
    return RequestContextV2Http(self._data)

prop request_id : str

The ID that API Gateway assigns to the API request.

Expand source code

@property
def request_id(self) -> str:
    """The ID that API Gateway assigns to the API request."""
    return self["requestContext"]["requestId"]

prop route_key : str

The selected route key.

Expand source code

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

prop stage : str

The deployment stage of the API request

Expand source code

@property
def stage(self) -> str:
    """The deployment stage of the API request"""
    return self["requestContext"]["stage"]

prop time : str

The CLF-formatted request time (dd/MMM/yyyy:HH:mm:ss +-hhmm).

Expand source code

@property
def time(self) -> str:
    """The CLF-formatted request time (dd/MMM/yyyy:HH:mm:ss +-hhmm)."""
    return self["requestContext"]["time"]

prop time_epoch : int

The Epoch-formatted request time.

Expand source code

@property
def time_epoch(self) -> int:
    """The Epoch-formatted request time."""
    return self["requestContext"]["timeEpoch"]

Inherited members

• DictWrapper:
  • get
  • raw_event

class CaseInsensitiveDict (data=None, **kwargs)

Case insensitive dict implementation. Assumes string keys only.

Expand source code

class CaseInsensitiveDict(dict):
    """Case insensitive dict implementation. Assumes string keys only."""

    def __init__(self, data=None, **kwargs):
        super().__init__()
        self.update(data, **kwargs)

    def get(self, k, default=None):
        return super().get(k.lower(), default)

    def pop(self, k):
        return super().pop(k.lower())

    def setdefault(self, k, default=None):
        return super().setdefault(k.lower(), default)

    def update(self, data=None, **kwargs):
        if data is not None:
            if isinstance(data, Mapping):
                data = data.items()
            super().update((k.lower(), v) for k, v in data)
        super().update((k.lower(), v) for k, v in kwargs)

    def __contains__(self, k):
        return super().__contains__(k.lower())

    def __delitem__(self, k):
        super().__delitem__(k.lower())

    def __eq__(self, other):
        if not isinstance(other, Mapping):
            return False
        if not isinstance(other, CaseInsensitiveDict):
            other = CaseInsensitiveDict(other)
        return super().__eq__(other)

    def __getitem__(self, k):
        return super().__getitem__(k.lower())

    def __setitem__(self, k, v):
        super().__setitem__(k.lower(), v)

Ancestors

• builtins.dict

Methods

def get(self, k, default=None)

Return the value for key if key is in the dictionary, else default.

def pop(self, k)

D.pop(k[,d]) -> v, remove specified key and return the corresponding value.

If the key is not found, return the default if given; otherwise, raise a KeyError.

def setdefault(self, k, default=None)

Insert key with a value of default if key is not in the dictionary.

Return the value for key if key is in the dictionary, else default.

def update(self, data=None, **kwargs)

D.update([E, ]**F) -> None. Update D from mapping/iterable E and F. If E is present and has a .keys() method, then does: for k in E.keys(): D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

class DictWrapper (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 DictWrapper(Mapping):
    """Provides a single read only access to a wrapper dict"""

    def __init__(self, data: dict[str, Any], json_deserializer: Callable | None = None):
        """
        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
        """
        self._data = data
        self._json_deserializer = json_deserializer or json.loads

    def __getitem__(self, key: str) -> Any:
        return self._data[key]

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, DictWrapper):
            return False

        return self._data == other._data

    def __iter__(self) -> Iterator:
        return iter(self._data)

    def __len__(self) -> int:
        return len(self._data)

    def __str__(self) -> str:
        return str(self._str_helper())

    def _str_helper(self) -> dict[str, Any]:
        """
        Recursively get a Dictionary of DictWrapper properties primarily
        for use by __str__ for debugging purposes.

        Will remove "raw_event" properties, and any defined by the Data Class
        `_sensitive_properties` list field.
        This should be used in case where secrets, such as access keys, are
        stored in the Data Class but should not be logged out.
        """
        properties = self._properties()
        sensitive_properties = ["raw_event"]
        if hasattr(self, "_sensitive_properties"):
            sensitive_properties.extend(self._sensitive_properties)  # pyright: ignore

        result: dict[str, Any] = {}
        for property_key in properties:
            if property_key in sensitive_properties:
                result[property_key] = "[SENSITIVE]"
            else:
                try:
                    property_value = getattr(self, property_key)
                    result[property_key] = property_value

                    # Checks whether the class is a subclass of the parent class to perform a recursive operation.
                    if issubclass(property_value.__class__, DictWrapper):
                        result[property_key] = property_value._str_helper()
                    # Checks if the key is a list and if it is a subclass of the parent class
                    elif isinstance(property_value, list):
                        for seq, item in enumerate(property_value):
                            if issubclass(item.__class__, DictWrapper):
                                result[property_key][seq] = item._str_helper()
                except Exception:
                    result[property_key] = "[Cannot be deserialized]"

        return result

    def _properties(self) -> list[str]:
        return [p for p in dir(self.__class__) if isinstance(getattr(self.__class__, p), property)]

    def get(self, key: str, default: Any | None = None) -> Any | None:
        return self._data.get(key, default)

    @property
    def raw_event(self) -> dict[str, Any]:
        """The original raw event dict"""
        return self._data

Ancestors

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

Subclasses

• ActiveMQEvent
• ActiveMQMessage
• ALBEventRequestContext
• APIGatewayAuthorizerEventV2
• APIGatewayAuthorizerRequestEvent
• APIGatewayAuthorizerTokenEvent
• APIGatewayEventAuthorizer
• RequestContextV2Authorizer
• RequestContextV2AuthorizerIam
• AppSyncAuthorizerEvent
• AppSyncAuthorizerEventRequestContext
• AppSyncIdentityCognito
• AppSyncIdentityIAM
• AppSyncResolverEvent
• AppSyncResolverEventInfo
• AWSConfigConfigurationChanged
• AWSConfigConfigurationItemChanged
• AWSConfigOversizedConfiguration
• AWSConfigOversizedConfigurationItemSummary
• AWSConfigRuleEvent
• AWSConfigScheduledNotification
• BedrockAgentInfo
• BedrockAgentProperty
• BedrockAgentRequestBody
• BedrockAgentRequestMedia
• CloudWatchAlarmConfiguration
• CloudWatchAlarmData
• CloudWatchAlarmEvent
• CloudWatchAlarmMetric
• CloudWatchAlarmMetricStat
• CloudWatchAlarmState
• CloudWatchDashboardCustomWidgetEvent
• CloudWatchWidgetContext
• TimeRange
• TimeZone
• CloudWatchLogsDecodedData
• CloudWatchLogsEvent
• CloudWatchLogsLogEvent
• CloudFormationCustomResourceEvent
• CodeDeployLifecycleHookEvent
• CodePipelineActionConfiguration
• CodePipelineArtifact
• CodePipelineArtifactCredentials
• CodePipelineConfiguration
• CodePipelineData
• CodePipelineEncryptionKey
• CodePipelineJobEvent
• CodePipelineLocation
• CodePipelineS3Location
• BaseTriggerEvent
• CallerContext
• ChallengeResult
• ClaimsAndScopeOverrideDetails
• ClaimsOverrideDetails
• CreateAuthChallengeTriggerEventRequest
• CreateAuthChallengeTriggerEventResponse
• CustomEmailSenderTriggerEventRequest
• CustomMessageTriggerEventRequest
• CustomMessageTriggerEventResponse
• CustomSMSSenderTriggerEventRequest
• DefineAuthChallengeTriggerEventRequest
• DefineAuthChallengeTriggerEventResponse
• GroupOverrideDetails
• PostAuthenticationTriggerEventRequest
• PostConfirmationTriggerEventRequest
• PreAuthenticationTriggerEventRequest
• PreSignUpTriggerEventRequest
• PreSignUpTriggerEventResponse
• PreTokenGenerationTriggerEventRequest
• PreTokenGenerationTriggerEventResponse
• PreTokenGenerationTriggerV2EventResponse
• TokenClaimsAndScopeOverrideDetails
• UserMigrationTriggerEventRequest
• UserMigrationTriggerEventResponse
• VerifyAuthChallengeResponseTriggerEventRequest
• VerifyAuthChallengeResponseTriggerEventResponse
• APIGatewayEventIdentity
• BaseProxyEvent
• BaseRequestContext
• BaseRequestContextV2
• RequestContextClientCert
• RequestContextV2Http
• ConnectContactFlowData
• ConnectContactFlowEndpoint
• ConnectContactFlowEvent
• ConnectContactFlowMediaStreamAudio
• ConnectContactFlowMediaStreamCustomer
• ConnectContactFlowMediaStreams
• ConnectContactFlowQueue
• DynamoDBRecord
• DynamoDBStreamEvent
• StreamRecord
• EventBridgeEvent
• KafkaEvent
• KafkaEventRecord
• KinesisFirehoseEvent
• KinesisFirehoseRecord
• KinesisFirehoseRecordMetadata
• KinesisStreamEvent
• KinesisStreamRecord
• KinesisStreamRecordPayload
• BasicProperties
• RabbitMQEvent
• RabbitMessage
• S3BatchOperationEvent
• S3BatchOperationJob
• S3BatchOperationTask
• S3Bucket
• S3Event
• S3EventBridgeNotificationDetail
• S3EventBridgeNotificationObject
• S3EventNotificationEventBridgeBucket
• S3EventRecord
• S3EventRecordGlacierEventData
• S3EventRecordGlacierRestoreEventData
• S3Identity
• S3Message
• S3Object
• S3RequestParameters
• S3ObjectConfiguration
• S3ObjectContext
• S3ObjectLambdaEvent
• S3ObjectSessionAttributes
• S3ObjectSessionContext
• S3ObjectSessionIssuer
• S3ObjectUserIdentity
• S3ObjectUserRequest
• SecretsManagerEvent
• SESEvent
• SESEventRecord
• SESMail
• SESMailCommonHeaders
• SESMailHeader
• SESMessage
• SESReceipt
• SESReceiptAction
• SESReceiptStatus
• SNSEvent
• SNSEventRecord
• SNSMessage
• SNSMessageAttribute
• SQSEvent
• SQSMessageAttribute
• SQSRecord
• SQSRecordAttributes
• vpcLatticeEventV2Identity
• vpcLatticeEventV2RequestContext

Instance variables

prop raw_event : dict[str, Any]

The original raw event dict

Expand source code

@property
def raw_event(self) -> dict[str, Any]:
    """The original raw event dict"""
    return self._data

Methods

def get(self, key: str, default: Any | None = None) ‑> typing.Any | None

D.get(k[,d]) -> D[k] if k in D, else d. d defaults to None.

class RequestContextClientCert (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 RequestContextClientCert(DictWrapper):
    @property
    def client_cert_pem(self) -> str:
        """Client certificate pem"""
        return self["clientCertPem"]

    @property
    def issuer_dn(self) -> str:
        """Issuer Distinguished Name"""
        return self["issuerDN"]

    @property
    def serial_number(self) -> str:
        """Unique serial number for client cert"""
        return self["serialNumber"]

    @property
    def subject_dn(self) -> str:
        """Subject Distinguished Name"""
        return self["subjectDN"]

    @property
    def validity_not_after(self) -> str:
        """Date when the cert is no longer valid

        eg: Aug  5 00:28:21 2120 GMT"""
        return self["validity"]["notAfter"]

    @property
    def validity_not_before(self) -> str:
        """Cert is not valid before this date

        eg: Aug 29 00:28:21 2020 GMT"""
        return self["validity"]["notBefore"]

Ancestors

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

Instance variables

prop client_cert_pem : str

Client certificate pem

Expand source code

@property
def client_cert_pem(self) -> str:
    """Client certificate pem"""
    return self["clientCertPem"]

prop issuer_dn : str

Issuer Distinguished Name

Expand source code

@property
def issuer_dn(self) -> str:
    """Issuer Distinguished Name"""
    return self["issuerDN"]

prop serial_number : str

Unique serial number for client cert

Expand source code

@property
def serial_number(self) -> str:
    """Unique serial number for client cert"""
    return self["serialNumber"]

prop subject_dn : str

Subject Distinguished Name

Expand source code

@property
def subject_dn(self) -> str:
    """Subject Distinguished Name"""
    return self["subjectDN"]

prop validity_not_after : str

Date when the cert is no longer valid

eg: Aug 5 00:28:21 2120 GMT

Expand source code

@property
def validity_not_after(self) -> str:
    """Date when the cert is no longer valid

    eg: Aug  5 00:28:21 2120 GMT"""
    return self["validity"]["notAfter"]

prop validity_not_before : str

Cert is not valid before this date

eg: Aug 29 00:28:21 2020 GMT

Expand source code

@property
def validity_not_before(self) -> str:
    """Cert is not valid before this date

    eg: Aug 29 00:28:21 2020 GMT"""
    return self["validity"]["notBefore"]

Inherited members

• DictWrapper:
  • get
  • raw_event

class RequestContextV2Http (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 RequestContextV2Http(DictWrapper):
    @property
    def method(self) -> str:
        return self["requestContext"]["http"]["method"]

    @property
    def path(self) -> str:
        return self["requestContext"]["http"]["path"]

    @property
    def protocol(self) -> str:
        """The request protocol, for example, HTTP/1.1."""
        return self["requestContext"]["http"]["protocol"]

    @property
    def source_ip(self) -> str:
        """The source IP address of the TCP connection making the request to API Gateway."""
        return self["requestContext"]["http"]["sourceIp"]

    @property
    def user_agent(self) -> str:
        """The User Agent of the API caller."""
        return self["requestContext"]["http"]["userAgent"]

Ancestors

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

Instance variables

prop method : str

Expand source code

@property
def method(self) -> str:
    return self["requestContext"]["http"]["method"]

prop path : str

Expand source code

@property
def path(self) -> str:
    return self["requestContext"]["http"]["path"]

prop protocol : str

The request protocol, for example, HTTP/1.1.

Expand source code

@property
def protocol(self) -> str:
    """The request protocol, for example, HTTP/1.1."""
    return self["requestContext"]["http"]["protocol"]

prop source_ip : str

The source IP address of the TCP connection making the request to API Gateway.

Expand source code

@property
def source_ip(self) -> str:
    """The source IP address of the TCP connection making the request to API Gateway."""
    return self["requestContext"]["http"]["sourceIp"]

prop user_agent : str

The User Agent of the API caller.

Expand source code

@property
def user_agent(self) -> str:
    """The User Agent of the API caller."""
    return self["requestContext"]["http"]["userAgent"]

Inherited members

• DictWrapper:
  • get
  • raw_event