feat(parser): add support for SQS-wrapped S3 event notifications

aws_lambda_powertools/utilities/parser/models/__init__.py

@@ -50,6 +50,10 @@
     S3Model,
     S3RecordModel,
 )
+from .s3_event_notification import (
+    SqsS3EventNotificationModel,
+    SqsS3EventNotificationRecordModel,
+)
 from .s3_object_event import (
     S3ObjectConfiguration,
     S3ObjectContext,
@@ -130,6 +134,8 @@
     "SqsRecordModel",
     "SqsMsgAttributeModel",
     "SqsAttributesModel",
+    "SqsS3EventNotificationModel",
+    "SqsS3EventNotificationRecordModel",
     "APIGatewayProxyEventModel",
     "APIGatewayEventRequestContext",
     "APIGatewayEventAuthorizer",

aws_lambda_powertools/utilities/parser/models/s3_event_notification.py

@@ -0,0 +1,14 @@
+from typing import List
+
+from pydantic import Json
+
+from aws_lambda_powertools.utilities.parser.models.s3 import S3Model
+from aws_lambda_powertools.utilities.parser.models.sqs import SqsModel, SqsRecordModel
+
+
+class SqsS3EventNotificationRecordModel(SqsRecordModel):
+    body: Json[S3Model]
+
+
+class SqsS3EventNotificationModel(SqsModel):
+    Records: List[SqsS3EventNotificationRecordModel]

aws_lambda_powertools/utilities/parser/models/sqs.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Dict, List, Optional, Type, Union
+from typing import Dict, List, Optional, Sequence, Type, Union
 
 from pydantic import BaseModel
 
@@ -63,4 +63,4 @@ class SqsRecordModel(BaseModel):
 
 
 class SqsModel(BaseModel):
-    Records: List[SqsRecordModel]
+    Records: Sequence[SqsRecordModel]

docs/utilities/parser.md

@@ -175,6 +175,7 @@ Parser comes with the following built-in models:
 | **LambdaFunctionUrlModel**              | Lambda Event Source payload for Lambda Function URL payload                  |
 | **KafkaSelfManagedEventModel**          | Lambda Event Source payload for self managed Kafka payload                   |
 | **KafkaMskEventModel**                  | Lambda Event Source payload for AWS MSK payload                              |
+| **SqsS3EventNotificationModel**         | Lambda Event Source payload for SQS-wrapped S3 event notifications           |
 
 #### Extending built-in models
 

tests/functional/parser/test_s3.py

@@ -6,8 +6,7 @@
 from tests.functional.utils import load_event
 
 
-@event_parser(model=S3Model)
-def handle_s3(event: S3Model, _: LambdaContext):
+def assert_s3(event: S3Model):
     records = list(event.Records)
     assert len(records) == 1
     record: S3RecordModel = records[0]
@@ -41,6 +40,11 @@ def handle_s3(event: S3Model, _: LambdaContext):
     assert record.glacierEventData is None
 
 
+@event_parser(model=S3Model)
+def handle_s3(event: S3Model, _: LambdaContext):
+    assert_s3(event)
+
+
 @event_parser(model=S3Model)
 def handle_s3_glacier(event: S3Model, _: LambdaContext):
     records = list(event.Records)

tests/functional/parser/test_sqs_s3_event_notification.py

@@ -0,0 +1,43 @@
+import pytest
+
+from aws_lambda_powertools.utilities.parser import ValidationError, event_parser
+from aws_lambda_powertools.utilities.parser.models import SqsS3EventNotificationModel
+from aws_lambda_powertools.utilities.typing import LambdaContext
+from tests.functional.parser.test_s3 import assert_s3
+from tests.functional.utils import json_serialize, load_event
+
+
+@event_parser(model=SqsS3EventNotificationModel)
+def handle_sqs_json_body_containing_s3_notifications(event: SqsS3EventNotificationModel, _: LambdaContext):
+    return event
+
+
+def test_handle_sqs_json_body_containing_s3_notifications():
+    sqs_event_dict = load_event("sqsEvent.json")
+    s3_event_notification_dict = load_event("s3Event.json")
+    for record in sqs_event_dict["Records"]:
+        record["body"] = json_serialize(s3_event_notification_dict)
+
+    parsed_event: SqsS3EventNotificationModel = handle_sqs_json_body_containing_s3_notifications(
+        sqs_event_dict, LambdaContext()
+    )
+
+    assert len(parsed_event.Records) == 2
+    for parsed_sqs_record in parsed_event.Records:
+        assert_s3(parsed_sqs_record.body)
+
+
+def test_handle_sqs_body_invalid_json():
+    sqs_event_dict = load_event("sqsEvent.json")
+
+    with pytest.raises(ValidationError):
+        handle_sqs_json_body_containing_s3_notifications(sqs_event_dict, LambdaContext())
+
+
+def test_handle_sqs_json_body_containing_arbitrary_json():
+    sqs_event_dict = load_event("sqsEvent.json")
+    for record in sqs_event_dict["Records"]:
+        record["body"] = json_serialize({"foo": "bar"})
+
+    with pytest.raises(ValidationError):
+        handle_sqs_json_body_containing_s3_notifications(sqs_event_dict, LambdaContext())