feat(nested-event-sources): Generic unwrapping of event source data

aws_lambda_powertools/utilities/data_classes/cloud_watch_logs_event.py

@@ -2,10 +2,10 @@
 import zlib
 from typing import Dict, List, Optional
 
-from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper, EventWrapper
 
 
-class CloudWatchLogsLogEvent(DictWrapper):
+class CloudWatchLogsLogEvent(EventWrapper):
     @property
     def get_id(self) -> str:
         """The ID property is a unique identifier for every log event."""
@@ -72,7 +72,7 @@ def log_events(self) -> List[CloudWatchLogsLogEvent]:
         return [CloudWatchLogsLogEvent(i) for i in self["logEvents"]]
 
 
-class CloudWatchLogsEvent(DictWrapper):
+class CloudWatchLogsEvent(EventWrapper):
     """CloudWatch Logs log stream event
 
     You can use a Lambda function to monitor and analyze logs from an Amazon CloudWatch Logs log stream.

aws_lambda_powertools/utilities/data_classes/common.py

@@ -2,7 +2,7 @@
 import json
 from collections.abc import Mapping
 from functools import cached_property
-from typing import Any, Callable, Dict, Iterator, List, Optional, overload
+from typing import Any, Callable, Dict, Iterator, List, Optional, Type, TypeVar, overload
 
 from aws_lambda_powertools.shared.headers_serializer import BaseHeadersSerializer
 from aws_lambda_powertools.utilities.data_classes.shared_functions import (
@@ -95,6 +95,52 @@ def raw_event(self) -> Dict[str, Any]:
         return self._data
 
 
+class EventWrapper(DictWrapper):
+    NestedEvent = TypeVar("NestedEvent", bound=DictWrapper)
+
+    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
+        """
+        super().__init__(data, json_deserializer)
+
+    def nested_event_contents(self):
+        records = self.get("Records")
+        if records is None:
+            raise KeyError("No 'Records' key found in the event data.")
+        for record in records:
+            if not isinstance(record, dict):
+                raise TypeError(f"Expected 'Records' to be a dictionary, but got {type(record)}.")
+            body = record.get("body")
+            if body is not None:
+                yield body
+            else:
+                raise KeyError("No 'body' key found in the 'Records' dict.")
+
+    def decode_nested_events(self, nested_event_class: Type[NestedEvent], nested_event_content_deserializer=None):
+        if nested_event_content_deserializer is None:
+            nested_event_content_deserializer = self._json_deserializer
+
+        for content in self.nested_event_contents():
+            deserialized_data = nested_event_content_deserializer(content)
+            yield nested_event_class(deserialized_data)
+
+    def decode_nested_event(self, nested_event_class: Type[NestedEvent], nested_event_content_deserializer=None):
+        if nested_event_content_deserializer is None:
+            nested_event_content_deserializer = self._json_deserializer
+
+        for content in self.nested_event_contents():
+            deserialized_data = nested_event_content_deserializer(content)
+            return nested_event_class(deserialized_data)
+
+
 class BaseProxyEvent(DictWrapper):
     @property
     def headers(self) -> Dict[str, str]:
@@ -173,10 +219,12 @@ def http_method(self) -> str:
         return self["httpMethod"]
 
     @overload
-    def get_query_string_value(self, name: str, default_value: str) -> str: ...
+    def get_query_string_value(self, name: str, default_value: str) -> str:
+        ...
 
     @overload
-    def get_query_string_value(self, name: str, default_value: Optional[str] = None) -> Optional[str]: ...
+    def get_query_string_value(self, name: str, default_value: Optional[str] = None) -> Optional[str]:
+        ...
 
     def get_query_string_value(self, name: str, default_value: Optional[str] = None) -> Optional[str]:
         """Get query string value by name
@@ -229,15 +277,17 @@ def get_header_value(
         name: str,
         default_value: str,
         case_sensitive: bool = False,
-    ) -> str: ...
+    ) -> str:
+        ...
 
     @overload
     def get_header_value(
         self,
         name: str,
         default_value: Optional[str] = None,
         case_sensitive: bool = False,
-    ) -> Optional[str]: ...
+    ) -> Optional[str]:
+        ...
 
     def get_header_value(
         self,

aws_lambda_powertools/utilities/data_classes/event_bridge_event.py

@@ -1,9 +1,10 @@
+import json
 from typing import Any, Dict, List, Optional
 
-from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+from aws_lambda_powertools.utilities.data_classes.common import EventWrapper
 
 
-class EventBridgeEvent(DictWrapper):
+class EventBridgeEvent(EventWrapper):
     """Amazon EventBridge Event
 
     Documentation:
@@ -67,3 +68,6 @@ def detail(self) -> Dict[str, Any]:
     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"]
+
+    def nested_event_contents(self):
+        yield json.dumps(self.get("detail"))

aws_lambda_powertools/utilities/data_classes/kafka_event.py

@@ -75,15 +75,17 @@ def get_header_value(
         name: str,
         default_value: str,
         case_sensitive: bool = True,
-    ) -> str: ...
+    ) -> str:
+        ...
 
     @overload
     def get_header_value(
         self,
         name: str,
         default_value: Optional[str] = None,
         case_sensitive: bool = True,
-    ) -> Optional[str]: ...
+    ) -> Optional[str]:
+        ...
 
     def get_header_value(
         self,

aws_lambda_powertools/utilities/data_classes/kinesis_firehose_event.py

@@ -7,7 +7,7 @@
 
 from typing_extensions import Literal
 
-from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper, EventWrapper
 
 
 @dataclass(repr=False, order=False, frozen=True)
@@ -208,7 +208,7 @@ def subsequence_number(self) -> int:
         return self._metadata["subsequenceNumber"]
 
 
-class KinesisFirehoseRecord(DictWrapper):
+class KinesisFirehoseRecord(EventWrapper):
     @property
     def approximate_arrival_timestamp(self) -> int:
         """The approximate time that the record was inserted into the delivery stream"""
@@ -271,7 +271,7 @@ def build_data_transformation_response(
         )
 
 
-class KinesisFirehoseEvent(DictWrapper):
+class KinesisFirehoseEvent(EventWrapper):
     """Kinesis Data Firehose event
 
     Documentation:
@@ -303,3 +303,9 @@ def region(self) -> str:
     def records(self) -> Iterator[KinesisFirehoseRecord]:
         for record in self["records"]:
             yield KinesisFirehoseRecord(data=record, json_deserializer=self._json_deserializer)
+
+    def nested_event_contents(self):
+        for record in self.get("records"):
+            body = record.get("data")
+            body = base64.b64decode(body).decode("utf-8")
+            yield body

aws_lambda_powertools/utilities/data_classes/kinesis_stream_event.py

@@ -6,7 +6,7 @@
 from aws_lambda_powertools.utilities.data_classes.cloud_watch_logs_event import (
     CloudWatchLogsDecodedData,
 )
-from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper, EventWrapper
 
 
 class KinesisStreamRecordPayload(DictWrapper):
@@ -95,7 +95,7 @@ def kinesis(self) -> KinesisStreamRecordPayload:
         return KinesisStreamRecordPayload(self._data)
 
 
-class KinesisStreamEvent(DictWrapper):
+class KinesisStreamEvent(EventWrapper):
     """Kinesis stream event
 
     Documentation:

aws_lambda_powertools/utilities/data_classes/s3_event.py

@@ -1,7 +1,7 @@
 from typing import Dict, Iterator, Optional
 from urllib.parse import unquote_plus
 
-from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper, EventWrapper
 from aws_lambda_powertools.utilities.data_classes.event_bridge_event import (
     EventBridgeEvent,
 )
@@ -19,13 +19,13 @@ def source_ip_address(self) -> str:
         return self["requestParameters"]["sourceIPAddress"]
 
 
-class S3EventNotificationEventBridgeBucket(DictWrapper):
+class S3EventNotificationEventBridgeBucket(EventWrapper):
     @property
     def name(self) -> str:
         return self["name"]
 
 
-class S3EventBridgeNotificationObject(DictWrapper):
+class S3EventBridgeNotificationObject(EventWrapper):
     @property
     def key(self) -> str:
         """Object key"""
@@ -52,7 +52,7 @@ def sequencer(self) -> str:
         return self["sequencer"]
 
 
-class S3EventBridgeNotificationDetail(DictWrapper):
+class S3EventBridgeNotificationDetail(EventWrapper):
     @property
     def version(self) -> str:
         """Get the detail version"""
@@ -151,7 +151,7 @@ def detail(self) -> S3EventBridgeNotificationDetail:  # type: ignore[override]
         return S3EventBridgeNotificationDetail(self["detail"])
 
 
-class S3Bucket(DictWrapper):
+class S3Bucket(EventWrapper):
     @property
     def name(self) -> str:
         return self["s3"]["bucket"]["name"]
@@ -165,7 +165,7 @@ def arn(self) -> str:
         return self["s3"]["bucket"]["arn"]
 
 
-class S3Object(DictWrapper):
+class S3Object(EventWrapper):
     @property
     def key(self) -> str:
         """Object key"""
@@ -194,7 +194,7 @@ def sequencer(self) -> str:
         return self["s3"]["object"]["sequencer"]
 
 
-class S3Message(DictWrapper):
+class S3Message(EventWrapper):
     @property
     def s3_schema_version(self) -> str:
         return self["s3"]["s3SchemaVersion"]
@@ -237,7 +237,7 @@ def restore_event_data(self) -> S3EventRecordGlacierRestoreEventData:
         return S3EventRecordGlacierRestoreEventData(self._data)
 
 
-class S3EventRecord(DictWrapper):
+class S3EventRecord(EventWrapper):
     @property
     def event_version(self) -> str:
         """The eventVersion key value contains a major and minor version in the form <major>.<minor>."""
@@ -293,7 +293,7 @@ def glacier_event_data(self) -> Optional[S3EventRecordGlacierEventData]:
         return None if item is None else S3EventRecordGlacierEventData(item)
 
 
-class S3Event(DictWrapper):
+class S3Event(EventWrapper):
     """S3 event notification
 
     Documentation:

aws_lambda_powertools/utilities/data_classes/s3_object_event.py

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

aws_lambda_powertools/utilities/data_classes/ses_event.py

@@ -1,9 +1,9 @@
 from typing import Iterator, List, Optional
 
-from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+from aws_lambda_powertools.utilities.data_classes.common import EventWrapper
 
 
-class SESMailHeader(DictWrapper):
+class SESMailHeader(EventWrapper):
     @property
     def name(self) -> str:
         return self["name"]
@@ -13,7 +13,7 @@ def value(self) -> str:
         return self["value"]
 
 
-class SESMailCommonHeaders(DictWrapper):
+class SESMailCommonHeaders(EventWrapper):
     @property
     def return_path(self) -> str:
         """The values in the Return-Path header of the email."""
@@ -66,7 +66,7 @@ def reply_to(self) -> Optional[List[str]]:
         return self.get("replyTo")
 
 
-class SESMail(DictWrapper):
+class SESMail(EventWrapper):
     @property
     def timestamp(self) -> str:
         """String that contains the time at which the email was received, in ISO8601 format."""
@@ -111,7 +111,7 @@ def common_headers(self) -> SESMailCommonHeaders:
         return SESMailCommonHeaders(self["commonHeaders"])
 
 
-class SESReceiptStatus(DictWrapper):
+class SESReceiptStatus(EventWrapper):
     @property
     def status(self) -> str:
         """Receipt status
@@ -120,7 +120,7 @@ def status(self) -> str:
         return str(self["status"])
 
 
-class SESReceiptAction(DictWrapper):
+class SESReceiptAction(EventWrapper):
     @property
     def get_type(self) -> str:
         """String that indicates the type of action that was executed.
@@ -149,7 +149,7 @@ def invocation_type(self) -> str:
         return self["invocationType"]
 
 
-class SESReceipt(DictWrapper):
+class SESReceipt(EventWrapper):
     @property
     def timestamp(self) -> str:
         """String that specifies the date and time at which the action was triggered, in ISO 8601 format."""
@@ -207,7 +207,7 @@ def action(self) -> SESReceiptAction:
         return SESReceiptAction(self["action"])
 
 
-class SESMessage(DictWrapper):
+class SESMessage(EventWrapper):
     @property
     def mail(self) -> SESMail:
         return SESMail(self["ses"]["mail"])
@@ -217,7 +217,7 @@ def receipt(self) -> SESReceipt:
         return SESReceipt(self["ses"]["receipt"])
 
 
-class SESEventRecord(DictWrapper):
+class SESEventRecord(EventWrapper):
     @property
     def event_source(self) -> str:
         """The AWS service from which the SES event record originated. For SES, this is aws:ses"""
@@ -233,7 +233,7 @@ def ses(self) -> SESMessage:
         return SESMessage(self._data)
 
 
-class SESEvent(DictWrapper):
+class SESEvent(EventWrapper):
     """Amazon SES to receive message event trigger
 
     NOTE: There is a 30-second timeout on RequestResponse invocations.
@@ -260,3 +260,8 @@ def mail(self) -> SESMail:
     @property
     def receipt(self) -> SESReceipt:
         return self.record.ses.receipt
+
+    def nested_event_contents(self):
+        for record in self.get("Records"):
+            body = record.get("ses")
+            yield body

aws_lambda_powertools/utilities/data_classes/sns_event.py

@@ -1,6 +1,6 @@
 from typing import Dict, Iterator
 
-from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper, EventWrapper
 
 
 class SNSMessageAttribute(DictWrapper):
@@ -16,7 +16,7 @@ def value(self) -> str:
         return self["Value"]
 
 
-class SNSMessage(DictWrapper):
+class SNSMessage(EventWrapper):
     @property
     def signature_version(self) -> str:
         """Version of the Amazon SNS signature used."""
@@ -78,8 +78,11 @@ def subject(self) -> str:
         """The Subject parameter specified when the notification was published to the topic."""
         return self["Subject"]
 
+    def nested_event_contents(self):
+        yield self.message
 
-class SNSEventRecord(DictWrapper):
+
+class SNSEventRecord(EventWrapper):
     @property
     def event_version(self) -> str:
         """Event version"""
@@ -99,7 +102,7 @@ def sns(self) -> SNSMessage:
         return SNSMessage(self._data["Sns"])
 
 
-class SNSEvent(DictWrapper):
+class SNSEvent(EventWrapper):
     """SNS Event
 
     Documentation:
@@ -121,3 +124,8 @@ def record(self) -> SNSEventRecord:
     def sns_message(self) -> str:
         """Return the message for the first sns event record"""
         return self.record.sns.message
+
+    def nested_event_contents(self):
+        for record in self.get("Records"):
+            body = record.get("Sns").get("Message")
+            yield body

aws_lambda_powertools/utilities/data_classes/sqs_event.py

@@ -2,7 +2,7 @@
 from typing import Any, Dict, ItemsView, Iterator, Optional, Type, TypeVar
 
 from aws_lambda_powertools.utilities.data_classes import S3Event
-from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper, EventWrapper
 from aws_lambda_powertools.utilities.data_classes.sns_event import SNSMessage
 
 
@@ -93,10 +93,10 @@ def items(self) -> ItemsView[str, SQSMessageAttribute]:  # type: ignore
         return {k: SQSMessageAttribute(v) for k, v in super().items()}.items()  # type: ignore
 
 
-class SQSRecord(DictWrapper):
+class SQSRecord(EventWrapper):
     """An Amazon SQS message"""
 
-    NestedEvent = TypeVar("NestedEvent", bound=DictWrapper)
+    NestedEvent = TypeVar("NestedEvent", bound=EventWrapper)
 
     @property
     def message_id(self) -> str:
@@ -245,7 +245,7 @@ def _decode_nested_event(self, nested_event_class: Type[NestedEvent]) -> NestedE
         return nested_event_class(self.json_body)
 
 
-class SQSEvent(DictWrapper):
+class SQSEvent(EventWrapper):
     """SQS Event
 
     Documentation:

tests/unit/data_classes/test_nested_events.py

@@ -0,0 +1,127 @@
+import aws_lambda_powertools.utilities.data_classes.nested_test_events as nested_test_events
+from aws_lambda_powertools.utilities.data_classes import (
+    EventBridgeEvent,
+    KinesisFirehoseEvent,
+    S3Event,
+    SNSEvent,
+    SQSEvent,
+)
+from aws_lambda_powertools.utilities.data_classes.s3_event import S3EventBridgeNotificationDetail
+from aws_lambda_powertools.utilities.data_classes.ses_event import SESEventRecord
+from aws_lambda_powertools.utilities.data_classes.sns_event import SNSMessage
+
+
+def test_sqs_s3():  # sqs(s3)
+    raw_event = nested_test_events.sqs_s3_event
+    parsed_event = SQSEvent(raw_event)
+
+    records = list(parsed_event.records)
+    assert len(records) == 1
+    record = records[0]
+    record_raw = raw_event["Records"][0]
+    assert record.aws_region == record_raw["awsRegion"]
+
+    s3_event = parsed_event.decode_nested_events(S3Event)
+    for rec in s3_event:
+        assert rec.bucket_name == "sqs-s3-unwrap-bucket"
+
+
+def test_sqs_s3_single():  # sqs(s3)
+    raw_event = nested_test_events.sqs_s3_event
+    parsed_event = SQSEvent(raw_event)
+
+    s3_event = parsed_event.decode_nested_event(S3Event)
+    assert s3_event.bucket_name == "sqs-s3-unwrap-bucket"
+
+
+def test_sqs_sns():  # sqs(sns)
+    raw_event = nested_test_events.sqs_sns_event
+    parsed_event = SQSEvent(raw_event)
+
+    sns_event = parsed_event.decode_nested_events(SNSMessage)
+    for rec in sns_event:
+        assert rec.message == "From SNS"
+
+
+def test_sns_s3():  # sns(s3)
+    raw_event = nested_test_events.sns_s3_event
+    parsed_event = SNSEvent(raw_event)
+
+    s3_event = parsed_event.decode_nested_events(S3Event)
+    for rec in s3_event:
+        assert rec.bucket_name == "s3-sns-unwrap-bucket"
+
+
+def test_sqs_s3_multiple_events():  # sqs(s3, s3)
+    raw_event = nested_test_events.sqs_s3_multi_event
+    parsed_event = SQSEvent(raw_event)
+
+    s3_event = parsed_event.decode_nested_events(S3Event)
+    for idx, rec in enumerate(s3_event):
+        assert rec.bucket_name == "example-bucket"+str(idx)
+
+
+def test_sqs_sns_s3_direct():  # sqs(sns(s3))
+    raw_event = nested_test_events.sqs_sns_s3_event
+    parsed_event = SQSEvent(raw_event)
+
+    sns_event = parsed_event.decode_nested_event(SNSMessage)
+    s3_event = sns_event.decode_nested_event(S3Event)
+    assert s3_event.bucket_name == "unwraptestevents-bucket"
+
+
+def test_sqs_sns_s3():  # sqs(sns(s3))
+    raw_event = nested_test_events.sqs_sns_s3_event
+    parsed_event = SQSEvent(raw_event)
+
+    sns_event = parsed_event.decode_nested_events(SNSMessage)
+    for rec in sns_event:
+        s3_event = rec.decode_nested_events(S3Event)
+        for r in s3_event:
+            assert r.bucket_name == "unwraptestevents-bucket"
+
+
+def test_sns_ses():  # sns(ses)
+    raw_event = nested_test_events.sns_ses_event
+    parsed_event = SNSEvent(raw_event)
+    ses_event = parsed_event.decode_nested_events(SESEventRecord)
+    for rec in ses_event:
+        assert rec.get("mail").get("source") == "jsmith@amazon.com"
+        # print('rec:', rec.mail) #but can't do rec.mail bc no "SES" key
+
+
+def test_eb_s3():  # eventbridge(s3)
+    raw_event = nested_test_events.eb_s3_event
+    parsed_event = EventBridgeEvent(raw_event)
+    s3_event = parsed_event.decode_nested_events(S3EventBridgeNotificationDetail)
+    for rec in s3_event:
+        assert rec.bucket.name == "s3-eb-unwrap-sourcebucket"
+
+
+def test_sqs_eb_s3():  # sqs(eventbridge(s3))
+    raw_event = nested_test_events.sqs_eb_s3_event
+    parsed_event = SQSEvent(raw_event)
+
+    eb_event = parsed_event.decode_nested_events(EventBridgeEvent)
+    for rec in eb_event:
+        s3_event = rec.decode_nested_events(S3EventBridgeNotificationDetail)
+        for r in s3_event:
+            assert r.bucket.name == "s3-eb-unwrap-sourcebucket"
+
+
+def test_firehose_sns_event():  # firehose(sns)
+    raw_event = nested_test_events.firehose_sns_event
+    parsed_event = KinesisFirehoseEvent(raw_event)
+
+    sns_event = parsed_event.decode_nested_events(SNSMessage)  # TODO: add a flag to auto decode?
+    for rec in sns_event:
+        assert rec.message == "message from sns"
+
+
+def test_firehose_cw_event():  # firehose(cw)
+    raw_event = nested_test_events.firehose_cw_event
+    parsed_event = KinesisFirehoseEvent(raw_event)
+
+    # cw_event = parsed_event.decode_nested_events(CloudWatchLogsLogEvent) #TODO: gives back encrypted data and can't decode w b64
+    # for rec in cw_event:
+    #     print('type:', type(rec), rec)