Module aws_lambda_powertools.utilities.data_classes

Event Source Data Classes utility provides classes self-describing Lambda event sources.

"""
Event Source Data Classes utility provides classes self-describing Lambda event sources.
"""

from .alb_event import ALBEvent
from .api_gateway_proxy_event import APIGatewayProxyEvent, APIGatewayProxyEventV2
from .appsync_resolver_event import AppSyncResolverEvent
from .cloud_watch_logs_event import CloudWatchLogsEvent
from .connect_contact_flow_event import ConnectContactFlowEvent
from .dynamo_db_stream_event import DynamoDBStreamEvent
from .event_bridge_event import EventBridgeEvent
from .kinesis_stream_event import KinesisStreamEvent
from .s3_event import S3Event
from .ses_event import SESEvent
from .sns_event import SNSEvent
from .sqs_event import SQSEvent

__all__ = [
    "APIGatewayProxyEvent",
    "APIGatewayProxyEventV2",
    "AppSyncResolverEvent",
    "ALBEvent",
    "CloudWatchLogsEvent",
    "ConnectContactFlowEvent",
    "DynamoDBStreamEvent",
    "EventBridgeEvent",
    "KinesisStreamEvent",
    "S3Event",
    "SESEvent",
    "SNSEvent",
    "SQSEvent",
]

Sub-modules

aws_lambda_powertools.utilities.data_classes.alb_event

aws_lambda_powertools.utilities.data_classes.api_gateway_proxy_event

aws_lambda_powertools.utilities.data_classes.appsync

aws_lambda_powertools.utilities.data_classes.appsync_resolver_event

aws_lambda_powertools.utilities.data_classes.cloud_watch_logs_event

aws_lambda_powertools.utilities.data_classes.cognito_user_pool_event

aws_lambda_powertools.utilities.data_classes.common

aws_lambda_powertools.utilities.data_classes.connect_contact_flow_event

aws_lambda_powertools.utilities.data_classes.dynamo_db_stream_event

aws_lambda_powertools.utilities.data_classes.event_bridge_event

aws_lambda_powertools.utilities.data_classes.kinesis_stream_event

aws_lambda_powertools.utilities.data_classes.s3_event

aws_lambda_powertools.utilities.data_classes.s3_object_event

aws_lambda_powertools.utilities.data_classes.ses_event

aws_lambda_powertools.utilities.data_classes.sns_event

aws_lambda_powertools.utilities.data_classes.sqs_event

Classes

class ALBEvent (data: Dict[str, Any])

Application load balancer event

Documentation:

• https://docs.aws.amazon.com/lambda/latest/dg/services-alb.html

Expand source code

class ALBEvent(BaseProxyEvent):
    """Application load balancer event

    Documentation:
    --------------
    - https://docs.aws.amazon.com/lambda/latest/dg/services-alb.html
    """

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

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

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

    @property
    def multi_value_query_string_parameters(self) -> Optional[Dict[str, List[str]]]:
        return self.get("multiValueQueryStringParameters")

    @property
    def multi_value_headers(self) -> Optional[Dict[str, List[str]]]:
        return self.get("multiValueHeaders")

Ancestors

• BaseProxyEvent
• DictWrapper

Instance variables

var http_method : str

Expand source code

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

var multi_value_headers : Union[Dict[str, List[str]], NoneType]

Expand source code

@property
def multi_value_headers(self) -> Optional[Dict[str, List[str]]]:
    return self.get("multiValueHeaders")

var multi_value_query_string_parameters : Union[Dict[str, List[str]], NoneType]

Expand source code

@property
def multi_value_query_string_parameters(self) -> Optional[Dict[str, List[str]]]:
    return self.get("multiValueQueryStringParameters")

var path : str

Expand source code

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

var request_context : ALBEventRequestContext

Expand source code

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

Inherited members

• BaseProxyEvent:
  • get_header_value
  • get_query_string_value
  • raw_event

class APIGatewayProxyEvent (data: Dict[str, Any])

AWS Lambda proxy V1

Documentation:

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

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

    @property
    def multi_value_headers(self) -> Dict[str, List[str]]:
        return self["multiValueHeaders"]

    @property
    def multi_value_query_string_parameters(self) -> Optional[Dict[str, List[str]]]:
        return self.get("multiValueQueryStringParameters")

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

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

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

Ancestors

• BaseProxyEvent
• DictWrapper

Instance variables

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 multi_value_headers : Dict[str, List[str]]

Expand source code

@property
def multi_value_headers(self) -> Dict[str, List[str]]:
    return self["multiValueHeaders"]

var multi_value_query_string_parameters : Union[Dict[str, List[str]], NoneType]

Expand source code

@property
def multi_value_query_string_parameters(self) -> Optional[Dict[str, List[str]]]:
    return self.get("multiValueQueryStringParameters")

var path : str

Expand source code

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

var path_parameters : Union[Dict[str, str], NoneType]

Expand source code

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

var request_context : APIGatewayEventRequestContext

Expand source code

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

var resource : str

Expand source code

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

var stage_variables : Union[Dict[str, str], NoneType]

Expand source code

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

var version : str

Expand source code

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

Inherited members

• BaseProxyEvent:
  • get_header_value
  • get_query_string_value
  • raw_event

class APIGatewayProxyEventV2 (data: Dict[str, Any])

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

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) -> Optional[List[str]]:
        return self.get("cookies")

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

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

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

Ancestors

• BaseProxyEvent
• DictWrapper

Instance variables

var cookies : Union[List[str], NoneType]

Expand source code

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

var path_parameters : Union[Dict[str, str], NoneType]

Expand source code

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

var raw_path : str

Expand source code

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

var raw_query_string : str

Expand source code

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

var request_context : RequestContextV2

Expand source code

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

var route_key : str

Expand source code

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

var stage_variables : Union[Dict[str, str], NoneType]

Expand source code

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

var version : str

Expand source code

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

Inherited members

• BaseProxyEvent:
  • get_header_value
  • get_query_string_value
  • raw_event

class AppSyncResolverEvent (data: dict)

AppSync resolver event

NOTE: AppSync Resolver Events can come in various shapes this data class supports both Amplify GraphQL directive @function and Direct Lambda Resolver

Documentation:

• https://docs.aws.amazon.com/appsync/latest/devguide/resolver-context-reference.html
• https://docs.amplify.aws/cli/graphql-transformer/function#structure-of-the-function-event

Expand source code

class AppSyncResolverEvent(DictWrapper):
    """AppSync resolver event

    **NOTE:** AppSync Resolver Events can come in various shapes this data class
    supports both Amplify GraphQL directive @function and Direct Lambda Resolver

    Documentation:
    -------------
    - https://docs.aws.amazon.com/appsync/latest/devguide/resolver-context-reference.html
    - https://docs.amplify.aws/cli/graphql-transformer/function#structure-of-the-function-event
    """

    def __init__(self, data: dict):
        super().__init__(data)

        info: Optional[dict] = data.get("info")
        if not info:
            info = {"fieldName": self.get("fieldName"), "parentTypeName": self.get("typeName")}

        self._info = AppSyncResolverEventInfo(info)

    @property
    def type_name(self) -> str:
        """The name of the parent type for the field that is currently being resolved."""
        return self.info.parent_type_name

    @property
    def field_name(self) -> str:
        """The name of the field that is currently being resolved."""
        return self.info.field_name

    @property
    def arguments(self) -> Dict[str, Any]:
        """A map that contains all GraphQL arguments for this field."""
        return self["arguments"]

    @property
    def identity(self) -> Union[None, AppSyncIdentityIAM, AppSyncIdentityCognito]:
        """An object that contains information about the caller.

        Depending of the type of identify found:

        - API_KEY authorization - returns None
        - AWS_IAM authorization - returns AppSyncIdentityIAM
        - AMAZON_COGNITO_USER_POOLS authorization - returns AppSyncIdentityCognito
        """
        return get_identity_object(self.get("identity"))

    @property
    def source(self) -> Optional[Dict[str, Any]]:
        """A map that contains the resolution of the parent field."""
        return self.get("source")

    @property
    def request_headers(self) -> Dict[str, str]:
        """Request headers"""
        return self["request"]["headers"]

    @property
    def prev_result(self) -> Optional[Dict[str, Any]]:
        """It represents the result of whatever previous operation was executed in a pipeline resolver."""
        prev = self.get("prev")
        if not prev:
            return None
        return prev.get("result")

    @property
    def info(self) -> AppSyncResolverEventInfo:
        """The info section contains information about the GraphQL request."""
        return self._info

    @property
    def stash(self) -> Optional[dict]:
        """The stash is a map that is made available inside each resolver and function mapping template.
        The same stash instance lives through a single resolver execution. This means that you can use the
        stash to pass arbitrary data across request and response mapping templates, and across functions in
        a pipeline resolver."""
        return self.get("stash")

    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
        Returns
        -------
        str, optional
            Header value
        """
        return get_header_value(self.request_headers, name, default_value, case_sensitive)

Ancestors

• DictWrapper

Instance variables

var arguments : Dict[str, Any]

A map that contains all GraphQL arguments for this field.

Expand source code

@property
def arguments(self) -> Dict[str, Any]:
    """A map that contains all GraphQL arguments for this field."""
    return self["arguments"]

var field_name : str

The name of the field that is currently being resolved.

Expand source code

@property
def field_name(self) -> str:
    """The name of the field that is currently being resolved."""
    return self.info.field_name

var identity : Union[NoneType, AppSyncIdentityIAM, AppSyncIdentityCognito]

An object that contains information about the caller.

Depending of the type of identify found:

• API_KEY authorization - returns None
• AWS_IAM authorization - returns AppSyncIdentityIAM
• AMAZON_COGNITO_USER_POOLS authorization - returns AppSyncIdentityCognito

Expand source code

@property
def identity(self) -> Union[None, AppSyncIdentityIAM, AppSyncIdentityCognito]:
    """An object that contains information about the caller.

    Depending of the type of identify found:

    - API_KEY authorization - returns None
    - AWS_IAM authorization - returns AppSyncIdentityIAM
    - AMAZON_COGNITO_USER_POOLS authorization - returns AppSyncIdentityCognito
    """
    return get_identity_object(self.get("identity"))

var info : AppSyncResolverEventInfo

The info section contains information about the GraphQL request.

Expand source code

@property
def info(self) -> AppSyncResolverEventInfo:
    """The info section contains information about the GraphQL request."""
    return self._info

var prev_result : Union[Dict[str, Any], NoneType]

It represents the result of whatever previous operation was executed in a pipeline resolver.

Expand source code

@property
def prev_result(self) -> Optional[Dict[str, Any]]:
    """It represents the result of whatever previous operation was executed in a pipeline resolver."""
    prev = self.get("prev")
    if not prev:
        return None
    return prev.get("result")

var request_headers : Dict[str, str]

Request headers

Expand source code

@property
def request_headers(self) -> Dict[str, str]:
    """Request headers"""
    return self["request"]["headers"]

var source : Union[Dict[str, Any], NoneType]

A map that contains the resolution of the parent field.

Expand source code

@property
def source(self) -> Optional[Dict[str, Any]]:
    """A map that contains the resolution of the parent field."""
    return self.get("source")

var stash : Union[dict, NoneType]

The stash is a map that is made available inside each resolver and function mapping template. The same stash instance lives through a single resolver execution. This means that you can use the stash to pass arbitrary data across request and response mapping templates, and across functions in a pipeline resolver.

Expand source code

@property
def stash(self) -> Optional[dict]:
    """The stash is a map that is made available inside each resolver and function mapping template.
    The same stash instance lives through a single resolver execution. This means that you can use the
    stash to pass arbitrary data across request and response mapping templates, and across functions in
    a pipeline resolver."""
    return self.get("stash")

var type_name : str

The name of the parent type for the field that is currently being resolved.

Expand source code

@property
def type_name(self) -> str:
    """The name of the parent type for the field that is currently being resolved."""
    return self.info.parent_type_name

Methods

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

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

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
    Returns
    -------
    str, optional
        Header value
    """
    return get_header_value(self.request_headers, name, default_value, case_sensitive)

Inherited members

• DictWrapper:
  • raw_event

class CloudWatchLogsEvent (data: Dict[str, Any])

CloudWatch Logs log stream event

You can use a Lambda function to monitor and analyze logs from an Amazon CloudWatch Logs log stream.

Documentation:

• https://docs.aws.amazon.com/lambda/latest/dg/services-cloudwatchlogs.html

Expand source code

class CloudWatchLogsEvent(DictWrapper):
    """CloudWatch Logs log stream event

    You can use a Lambda function to monitor and analyze logs from an Amazon CloudWatch Logs log stream.

    Documentation:
    --------------
    - https://docs.aws.amazon.com/lambda/latest/dg/services-cloudwatchlogs.html
    """

    _decompressed_logs_data = None
    _json_logs_data = None

    @property
    def raw_logs_data(self) -> str:
        """The value of the `data` field is a Base64 encoded ZIP archive."""
        return self["awslogs"]["data"]

    @property
    def decompress_logs_data(self) -> bytes:
        """Decode and decompress log data"""
        if self._decompressed_logs_data is None:
            payload = base64.b64decode(self.raw_logs_data)
            self._decompressed_logs_data = zlib.decompress(payload, zlib.MAX_WBITS | 32)
        return self._decompressed_logs_data

    def parse_logs_data(self) -> CloudWatchLogsDecodedData:
        """Decode, decompress and parse json data as CloudWatchLogsDecodedData"""
        if self._json_logs_data is None:
            self._json_logs_data = json.loads(self.decompress_logs_data.decode("UTF-8"))
        return CloudWatchLogsDecodedData(self._json_logs_data)

Ancestors

• DictWrapper

Instance variables

var decompress_logs_data : bytes

Decode and decompress log data

Expand source code

@property
def decompress_logs_data(self) -> bytes:
    """Decode and decompress log data"""
    if self._decompressed_logs_data is None:
        payload = base64.b64decode(self.raw_logs_data)
        self._decompressed_logs_data = zlib.decompress(payload, zlib.MAX_WBITS | 32)
    return self._decompressed_logs_data

var raw_logs_data : str

The value of the data field is a Base64 encoded ZIP archive.

Expand source code

@property
def raw_logs_data(self) -> str:
    """The value of the `data` field is a Base64 encoded ZIP archive."""
    return self["awslogs"]["data"]

Methods

def parse_logs_data(self) ‑> CloudWatchLogsDecodedData

Decode, decompress and parse json data as CloudWatchLogsDecodedData

Expand source code

def parse_logs_data(self) -> CloudWatchLogsDecodedData:
    """Decode, decompress and parse json data as CloudWatchLogsDecodedData"""
    if self._json_logs_data is None:
        self._json_logs_data = json.loads(self.decompress_logs_data.decode("UTF-8"))
    return CloudWatchLogsDecodedData(self._json_logs_data)

Inherited members

• DictWrapper:
  • raw_event

class ConnectContactFlowEvent (data: Dict[str, Any])

Amazon Connect contact flow event

Documentation:

• https://docs.aws.amazon.com/connect/latest/adminguide/connect-lambda-functions.html

Expand source code

class ConnectContactFlowEvent(DictWrapper):
    """Amazon Connect contact flow event

    Documentation:
    -------------
    - https://docs.aws.amazon.com/connect/latest/adminguide/connect-lambda-functions.html
    """

    @property
    def contact_data(self) -> ConnectContactFlowData:
        """This is always passed by Amazon Connect for every contact. Some parameters are optional."""
        return ConnectContactFlowData(self["Details"]["ContactData"])

    @property
    def parameters(self) -> Dict[str, str]:
        """These are parameters specific to this call that were defined when you created the Lambda function."""
        return self["Details"]["Parameters"]

Ancestors

• DictWrapper

Instance variables

var contact_data : ConnectContactFlowData

This is always passed by Amazon Connect for every contact. Some parameters are optional.

Expand source code

@property
def contact_data(self) -> ConnectContactFlowData:
    """This is always passed by Amazon Connect for every contact. Some parameters are optional."""
    return ConnectContactFlowData(self["Details"]["ContactData"])

var parameters : Dict[str, str]

These are parameters specific to this call that were defined when you created the Lambda function.

Expand source code

@property
def parameters(self) -> Dict[str, str]:
    """These are parameters specific to this call that were defined when you created the Lambda function."""
    return self["Details"]["Parameters"]

Inherited members

• DictWrapper:
  • raw_event

class DynamoDBStreamEvent (data: Dict[str, Any])

Dynamo DB Stream Event

Documentation:

• https://docs.aws.amazon.com/lambda/latest/dg/with-ddb.html

Expand source code

class DynamoDBStreamEvent(DictWrapper):
    """Dynamo DB Stream Event

    Documentation:
    -------------
    - https://docs.aws.amazon.com/lambda/latest/dg/with-ddb.html
    """

    @property
    def records(self) -> Iterator[DynamoDBRecord]:
        for record in self["Records"]:
            yield DynamoDBRecord(record)

Ancestors

• DictWrapper

Instance variables

var records : Iterator[DynamoDBRecord]

Expand source code

@property
def records(self) -> Iterator[DynamoDBRecord]:
    for record in self["Records"]:
        yield DynamoDBRecord(record)

Inherited members

• DictWrapper:
  • raw_event

class EventBridgeEvent (data: Dict[str, Any])

Amazon EventBridge Event

Documentation:

• https://docs.aws.amazon.com/eventbridge/latest/userguide/aws-events.html

Expand source code

class EventBridgeEvent(DictWrapper):
    """Amazon EventBridge Event

    Documentation:
    --------------
    - https://docs.aws.amazon.com/eventbridge/latest/userguide/aws-events.html
    """

    @property
    def get_id(self) -> str:
        """A unique value is generated for every event. This can be helpful in tracing events as
        they move through rules to targets, and are processed."""
        # Note: this name conflicts with existing python builtins
        return self["id"]

    @property
    def version(self) -> str:
        """By default, this is set to 0 (zero) in all events."""
        return self["version"]

    @property
    def account(self) -> str:
        """The 12-digit number identifying an AWS account."""
        return self["account"]

    @property
    def time(self) -> str:
        """The event timestamp, which can be specified by the service originating the event.

        If the event spans a time interval, the service might choose to report the start time, so
        this value can be noticeably before the time the event is actually received.
        """
        return self["time"]

    @property
    def region(self) -> str:
        """Identifies the AWS region where the event originated."""
        return self["region"]

    @property
    def resources(self) -> List[str]:
        """This JSON array contains ARNs that identify resources that are involved in the event.
        Inclusion of these ARNs is at the discretion of the service."""
        return self["resources"]

    @property
    def source(self) -> str:
        """Identifies the service that sourced the event. All events sourced from within AWS begin with "aws." """
        return self["source"]

    @property
    def detail_type(self) -> str:
        """Identifies, in combination with the source field, the fields and values that appear in the detail field."""
        return self["detail-type"]

    @property
    def detail(self) -> Dict[str, Any]:
        """A JSON object, whose content is at the discretion of the service originating the event. """
        return self["detail"]

    @property
    def replay_name(self) -> Optional[str]:
        """Identifies whether the event is being replayed and what is the name of the replay."""
        return self["replay-name"]

Ancestors

• DictWrapper

Instance variables

var account : str

The 12-digit number identifying an AWS account.

Expand source code

@property
def account(self) -> str:
    """The 12-digit number identifying an AWS account."""
    return self["account"]

var detail : Dict[str, Any]

A JSON object, whose content is at the discretion of the service originating the event.

Expand source code

@property
def detail(self) -> Dict[str, Any]:
    """A JSON object, whose content is at the discretion of the service originating the event. """
    return self["detail"]

var detail_type : str

Identifies, in combination with the source field, the fields and values that appear in the detail field.

Expand source code

@property
def detail_type(self) -> str:
    """Identifies, in combination with the source field, the fields and values that appear in the detail field."""
    return self["detail-type"]

var get_id : str

A unique value is generated for every event. This can be helpful in tracing events as they move through rules to targets, and are processed.

Expand source code

@property
def get_id(self) -> str:
    """A unique value is generated for every event. This can be helpful in tracing events as
    they move through rules to targets, and are processed."""
    # Note: this name conflicts with existing python builtins
    return self["id"]

var region : str

Identifies the AWS region where the event originated.

Expand source code

@property
def region(self) -> str:
    """Identifies the AWS region where the event originated."""
    return self["region"]

var replay_name : Union[str, NoneType]

Identifies whether the event is being replayed and what is the name of the replay.

Expand source code

@property
def replay_name(self) -> Optional[str]:
    """Identifies whether the event is being replayed and what is the name of the replay."""
    return self["replay-name"]

var resources : List[str]

This JSON array contains ARNs that identify resources that are involved in the event. Inclusion of these ARNs is at the discretion of the service.

Expand source code

@property
def resources(self) -> List[str]:
    """This JSON array contains ARNs that identify resources that are involved in the event.
    Inclusion of these ARNs is at the discretion of the service."""
    return self["resources"]

var source : str

Identifies the service that sourced the event. All events sourced from within AWS begin with "aws."

Expand source code

@property
def source(self) -> str:
    """Identifies the service that sourced the event. All events sourced from within AWS begin with "aws." """
    return self["source"]

var time : str

The event timestamp, which can be specified by the service originating the event.

If the event spans a time interval, the service might choose to report the start time, so this value can be noticeably before the time the event is actually received.

Expand source code

@property
def time(self) -> str:
    """The event timestamp, which can be specified by the service originating the event.

    If the event spans a time interval, the service might choose to report the start time, so
    this value can be noticeably before the time the event is actually received.
    """
    return self["time"]

var version : str

By default, this is set to 0 (zero) in all events.

Expand source code

@property
def version(self) -> str:
    """By default, this is set to 0 (zero) in all events."""
    return self["version"]

Inherited members

• DictWrapper:
  • raw_event

class KinesisStreamEvent (data: Dict[str, Any])

Kinesis stream event

Documentation:

• https://docs.aws.amazon.com/lambda/latest/dg/with-kinesis.html

Expand source code

class KinesisStreamEvent(DictWrapper):
    """Kinesis stream event

    Documentation:
    --------------
    - https://docs.aws.amazon.com/lambda/latest/dg/with-kinesis.html
    """

    @property
    def records(self) -> Iterator[KinesisStreamRecord]:
        for record in self["Records"]:
            yield KinesisStreamRecord(record)

Ancestors

• DictWrapper

Instance variables

var records : Iterator[KinesisStreamRecord]

Expand source code

@property
def records(self) -> Iterator[KinesisStreamRecord]:
    for record in self["Records"]:
        yield KinesisStreamRecord(record)

Inherited members

• DictWrapper:
  • raw_event

class S3Event (data: Dict[str, Any])

S3 event notification

Documentation:

• https://docs.aws.amazon.com/lambda/latest/dg/with-s3.html
• https://docs.aws.amazon.com/AmazonS3/latest/dev/NotificationHowTo.html
• https://docs.aws.amazon.com/AmazonS3/latest/dev/notification-content-structure.html

Expand source code

class S3Event(DictWrapper):
    """S3 event notification

    Documentation:
    -------------
    - https://docs.aws.amazon.com/lambda/latest/dg/with-s3.html
    - https://docs.aws.amazon.com/AmazonS3/latest/dev/NotificationHowTo.html
    - https://docs.aws.amazon.com/AmazonS3/latest/dev/notification-content-structure.html
    """

    @property
    def records(self) -> Iterator[S3EventRecord]:
        for record in self["Records"]:
            yield S3EventRecord(record)

    @property
    def record(self) -> S3EventRecord:
        """Get the first s3 event record"""
        return next(self.records)

    @property
    def bucket_name(self) -> str:
        """Get the bucket name for the first s3 event record"""
        return self["Records"][0]["s3"]["bucket"]["name"]

    @property
    def object_key(self) -> str:
        """Get the object key for the first s3 event record and unquote plus"""
        return unquote_plus(self["Records"][0]["s3"]["object"]["key"])

Ancestors

• DictWrapper

Instance variables

var bucket_name : str

Get the bucket name for the first s3 event record

Expand source code

@property
def bucket_name(self) -> str:
    """Get the bucket name for the first s3 event record"""
    return self["Records"][0]["s3"]["bucket"]["name"]

var object_key : str

Get the object key for the first s3 event record and unquote plus

Expand source code

@property
def object_key(self) -> str:
    """Get the object key for the first s3 event record and unquote plus"""
    return unquote_plus(self["Records"][0]["s3"]["object"]["key"])

var record : S3EventRecord

Get the first s3 event record

Expand source code

@property
def record(self) -> S3EventRecord:
    """Get the first s3 event record"""
    return next(self.records)

var records : Iterator[S3EventRecord]

Expand source code

@property
def records(self) -> Iterator[S3EventRecord]:
    for record in self["Records"]:
        yield S3EventRecord(record)

Inherited members

• DictWrapper:
  • raw_event

class SESEvent (data: Dict[str, Any])

Amazon SES to receive message event trigger

NOTE: There is a 30-second timeout on RequestResponse invocations.

Documentation:

• https://docs.aws.amazon.com/lambda/latest/dg/services-ses.html
• https://docs.aws.amazon.com/ses/latest/DeveloperGuide/receiving-email-action-lambda.html

Expand source code

class SESEvent(DictWrapper):
    """Amazon SES to receive message event trigger

    NOTE: There is a 30-second timeout on RequestResponse invocations.

    Documentation:
    --------------
    - https://docs.aws.amazon.com/lambda/latest/dg/services-ses.html
    - https://docs.aws.amazon.com/ses/latest/DeveloperGuide/receiving-email-action-lambda.html
    """

    @property
    def records(self) -> Iterator[SESEventRecord]:
        for record in self["Records"]:
            yield SESEventRecord(record)

    @property
    def record(self) -> SESEventRecord:
        return next(self.records)

    @property
    def mail(self) -> SESMail:
        return self.record.ses.mail

    @property
    def receipt(self) -> SESReceipt:
        return self.record.ses.receipt

Ancestors

• DictWrapper

Instance variables

var mail : SESMail

Expand source code

@property
def mail(self) -> SESMail:
    return self.record.ses.mail

var receipt : SESReceipt

Expand source code

@property
def receipt(self) -> SESReceipt:
    return self.record.ses.receipt

var record : SESEventRecord

Expand source code

@property
def record(self) -> SESEventRecord:
    return next(self.records)

var records : Iterator[SESEventRecord]

Expand source code

@property
def records(self) -> Iterator[SESEventRecord]:
    for record in self["Records"]:
        yield SESEventRecord(record)

Inherited members

• DictWrapper:
  • raw_event

class SNSEvent (data: Dict[str, Any])

SNS Event

Documentation:

• https://docs.aws.amazon.com/lambda/latest/dg/with-sns.html

Expand source code

class SNSEvent(DictWrapper):
    """SNS Event

    Documentation:
    -------------
    - https://docs.aws.amazon.com/lambda/latest/dg/with-sns.html
    """

    @property
    def records(self) -> Iterator[SNSEventRecord]:
        for record in self["Records"]:
            yield SNSEventRecord(record)

    @property
    def record(self) -> SNSEventRecord:
        """Return the first SNS event record"""
        return next(self.records)

    @property
    def sns_message(self) -> str:
        """Return the message for the first sns event record"""
        return self.record.sns.message

Ancestors

• DictWrapper

Instance variables

var record : SNSEventRecord

Return the first SNS event record

Expand source code

@property
def record(self) -> SNSEventRecord:
    """Return the first SNS event record"""
    return next(self.records)

var records : Iterator[SNSEventRecord]

Expand source code

@property
def records(self) -> Iterator[SNSEventRecord]:
    for record in self["Records"]:
        yield SNSEventRecord(record)

var sns_message : str

Return the message for the first sns event record

Expand source code

@property
def sns_message(self) -> str:
    """Return the message for the first sns event record"""
    return self.record.sns.message

Inherited members

• DictWrapper:
  • raw_event

class SQSEvent (data: Dict[str, Any])

SQS Event

Documentation:

• https://docs.aws.amazon.com/lambda/latest/dg/with-sqs.html

Expand source code

class SQSEvent(DictWrapper):
    """SQS Event

    Documentation:
    --------------
    - https://docs.aws.amazon.com/lambda/latest/dg/with-sqs.html
    """

    @property
    def records(self) -> Iterator[SQSRecord]:
        for record in self["Records"]:
            yield SQSRecord(record)

Ancestors

• DictWrapper

Instance variables

var records : Iterator[SQSRecord]

Expand source code

@property
def records(self) -> Iterator[SQSRecord]:
    for record in self["Records"]:
        yield SQSRecord(record)

Inherited members

• DictWrapper:
  • raw_event