Module aws_lambda_powertools.utilities.data_classes.common

Classes

class APIGatewayEventIdentity (data: Dict[str, Any], json_deserializer: Optional[Callable] = 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) -> Optional[str]:
        return self["requestContext"]["identity"].get("accessKey")

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

    @property
    def api_key(self) -> Optional[str]:
        """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) -> Optional[str]:
        """The API key ID associated with an API request that requires an API key."""
        return self["requestContext"]["identity"].get("apiKeyId")

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

    @property
    def cognito_authentication_provider(self) -> Optional[str]:
        """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) -> Optional[str]:
        """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) -> Optional[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["requestContext"]["identity"].get("cognitoIdentityId")

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

    @property
    def principal_org_id(self) -> Optional[str]:
        """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) -> Optional[str]:
        """The principal identifier of the user making the request."""
        return self["requestContext"]["identity"].get("user")

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

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

    @property
    def client_cert(self) -> Optional[RequestContextClientCert]:
        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

Instance variables

var access_key : Optional[str]

Expand source code

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

var account_id : Optional[str]

The AWS account ID associated with the request.

Expand source code

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

var api_key : Optional[str]

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) -> Optional[str]:
    """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")

var api_key_id : Optional[str]

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

Expand source code

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

var caller : Optional[str]

The principal identifier of the caller making the request.

Expand source code

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

var client_cert : Optional[RequestContextClientCert]

Expand source code

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

var cognito_authentication_provider : Optional[str]

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) -> Optional[str]:
    """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")

var cognito_authentication_type : Optional[str]

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) -> Optional[str]:
    """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")

var cognito_identity_id : Optional[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) -> Optional[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["requestContext"]["identity"].get("cognitoIdentityId")

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

var principal_org_id : Optional[str]

The AWS organization ID.

Expand source code

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

var 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"]

var user : Optional[str]

The principal identifier of the user making the request.

Expand source code

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

var user_agent : Optional[str]

The User Agent of the API caller.

Expand source code

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

var user_arn : Optional[str]

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

Expand source code

@property
def user_arn(self) -> Optional[str]:
    """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: Optional[Callable] = 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 self.get("headers") or {}

    @property
    def query_string_parameters(self) -> Optional[Dict[str, str]]:
        return self.get("queryStringParameters")

    @property
    def is_base64_encoded(self) -> Optional[bool]:
        return self.get("isBase64Encoded")

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

    @property
    def json_body(self) -> Any:
        """Parses the submitted body as json"""
        if self._json_data is None:
            self._json_data = self._json_deserializer(self.decoded_body)
        return self._json_data

    @property
    def decoded_body(self) -> str:
        """Dynamically base64 decode body as a str"""
        body: str = self["body"]
        if self.is_base64_encoded:
            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"]

    def get_query_string_value(self, name: str, default_value: Optional[str] = None) -> Optional[str]:
        """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,
        )

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

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

    def get_header_value(
        self,
        name: str,
        default_value: Optional[str] = None,
        case_sensitive: Optional[bool] = False,
    ) -> Optional[str]:
        """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
        """
        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

Subclasses

• ALBEvent
• APIGatewayProxyEvent
• APIGatewayProxyEventV2
• BedrockAgentEvent
• VPCLatticeEventBase

Instance variables

var body : Optional[str]

Submitted body of the request as a string

Expand source code

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

var decoded_body : str

Dynamically base64 decode body as a str

Expand source code

@property
def decoded_body(self) -> str:
    """Dynamically base64 decode body as a str"""
    body: str = self["body"]
    if self.is_base64_encoded:
        return base64.b64decode(body.encode()).decode()
    return body

var headers : Dict[str, str]

Expand source code

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

var 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"]

var is_base64_encoded : Optional[bool]

Expand source code

@property
def is_base64_encoded(self) -> Optional[bool]:
    return self.get("isBase64Encoded")

var json_body : Any

Parses the submitted body as json

Expand source code

@property
def json_body(self) -> Any:
    """Parses the submitted body as json"""
    if self._json_data is None:
        self._json_data = self._json_deserializer(self.decoded_body)
    return self._json_data

var path : str

Expand source code

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

var query_string_parameters : Optional[Dict[str, str]]

Expand source code

@property
def query_string_parameters(self) -> Optional[Dict[str, str]]:
    return self.get("queryStringParameters")

Methods

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

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

Expand source code

def get_header_value(
    self,
    name: str,
    default_value: Optional[str] = None,
    case_sensitive: Optional[bool] = False,
) -> Optional[str]:
    """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
    """
    return get_header_value(
        headers=self.headers,
        name=name,
        default_value=default_value,
        case_sensitive=case_sensitive,
    )

def get_query_string_value(self, name: str, default_value: Optional[str] = None) ‑> Optional[str]

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

Expand source code

def get_query_string_value(self, name: str, default_value: Optional[str] = None) -> Optional[str]:
    """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 header_serializer(self) ‑> BaseHeadersSerializer

Expand source code

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

Inherited members

• DictWrapper:
  • get
  • raw_event

class BaseRequestContext (data: Dict[str, Any], json_deserializer: Optional[Callable] = 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) -> Optional[str]:
        """A domain name"""
        return self["requestContext"].get("domainName")

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

    @property
    def extended_request_id(self) -> Optional[str]:
        """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) -> Optional[str]:
        """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

Subclasses

• APIGatewayEventRequestContext

Instance variables

var 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"]

var 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"]

var domain_name : Optional[str]

A domain name

Expand source code

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

var domain_prefix : Optional[str]

Expand source code

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

var extended_request_id : Optional[str]

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) -> Optional[str]:
    """An automatically generated ID for the API call, which contains more useful information
    for debugging/troubleshooting."""
    return self["requestContext"].get("extendedRequestId")

var 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"]

var identity : APIGatewayEventIdentity

Expand source code

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

var path : str

Expand source code

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

var 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"]

var 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"]

var request_time : Optional[str]

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

Expand source code

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

var 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"]

var resource_id : str

Expand source code

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

var resource_path : str

Expand source code

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

var 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: Optional[Callable] = 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) -> Optional[RequestContextClientCert]:
        """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

Subclasses

• RequestContextV2

Instance variables

var 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"]

var 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"]

var authentication : Optional[RequestContextClientCert]

Optional when using mutual TLS authentication

Expand source code

@property
def authentication(self) -> Optional[RequestContextClientCert]:
    """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)

var domain_name : str

A domain name

Expand source code

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

var domain_prefix : str

Expand source code

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

var http : RequestContextV2Http

Expand source code

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

var 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"]

var route_key : str

The selected route key.

Expand source code

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

var 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"]

var 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"]

var 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 DictWrapper (data: Dict[str, Any], json_deserializer: Optional[Callable] = 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: Optional[Callable] = 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_data: Optional[Any] = None
        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: Optional[Any] = None) -> Optional[Any]:
        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

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
• CloudWatchDashboardCustomWidgetEvent
• CloudWatchWidgetContext
• TimeRange
• TimeZone
• CloudWatchLogsDecodedData
• CloudWatchLogsEvent
• CloudWatchLogsLogEvent
• CodePipelineActionConfiguration
• CodePipelineArtifact
• CodePipelineArtifactCredentials
• CodePipelineConfiguration
• CodePipelineData
• CodePipelineEncryptionKey
• CodePipelineJobEvent
• CodePipelineLocation
• CodePipelineS3Location
• BaseTriggerEvent
• CallerContext
• ChallengeResult
• ClaimsOverrideDetails
• CreateAuthChallengeTriggerEventRequest
• CreateAuthChallengeTriggerEventResponse
• CustomMessageTriggerEventRequest
• CustomMessageTriggerEventResponse
• DefineAuthChallengeTriggerEventRequest
• DefineAuthChallengeTriggerEventResponse
• GroupOverrideDetails
• PostAuthenticationTriggerEventRequest
• PostConfirmationTriggerEventRequest
• PreAuthenticationTriggerEventRequest
• PreSignUpTriggerEventRequest
• PreSignUpTriggerEventResponse
• PreTokenGenerationTriggerEventRequest
• PreTokenGenerationTriggerEventResponse
• 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
• 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

var 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: Optional[Any] = None) ‑> Optional[Any]

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

Expand source code

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

class RequestContextClientCert (data: Dict[str, Any], json_deserializer: Optional[Callable] = 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

Instance variables

var client_cert_pem : str

Client certificate pem

Expand source code

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

var issuer_dn : str

Issuer Distinguished Name

Expand source code

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

var 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"]

var subject_dn : str

Subject Distinguished Name

Expand source code

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

var 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"]

var 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: Optional[Callable] = 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

Instance variables

var method : str

Expand source code

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

var path : str

Expand source code

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

var 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"]

var 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"]

var 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