feat(event_sources): support for S3 Event Notifications through EventBridge

.flake8

@@ -3,9 +3,11 @@ exclude = docs, .eggs, setup.py, example, .aws-sam, .git, dist, *.md, *.yaml, ex
 ignore = E203, E266, W503, BLK100, W291, I004
 max-line-length = 120
 max-complexity = 15
+; flake8-builtins isn't honouring inline ignore (A003)
 per-file-ignores =
     tests/e2e/utils/data_builder/__init__.py:F401
     tests/e2e/utils/data_fetcher/__init__.py:F401
+    aws_lambda_powertools/utilities/data_classes/s3_event.py:A003
 
 [isort]
 multi_line_output = 3

aws_lambda_powertools/utilities/data_classes/__init__.py

@@ -16,7 +16,7 @@
 from .kinesis_firehose_event import KinesisFirehoseEvent
 from .kinesis_stream_event import KinesisStreamEvent
 from .lambda_function_url_event import LambdaFunctionUrlEvent
-from .s3_event import S3Event
+from .s3_event import S3Event, S3EventBridgeNotificationEvent
 from .ses_event import SESEvent
 from .sns_event import SNSEvent
 from .sqs_event import SQSEvent
@@ -37,6 +37,7 @@
     "KinesisStreamEvent",
     "LambdaFunctionUrlEvent",
     "S3Event",
+    "S3EventBridgeNotificationEvent",
     "SESEvent",
     "SNSEvent",
     "SQSEvent",

aws_lambda_powertools/utilities/data_classes/s3_event.py

@@ -2,6 +2,9 @@
 from urllib.parse import unquote_plus
 
 from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+from aws_lambda_powertools.utilities.data_classes.event_bridge_event import (
+    EventBridgeEvent,
+)
 
 
 class S3Identity(DictWrapper):
@@ -16,6 +19,138 @@ def source_ip_address(self) -> str:
         return self["requestParameters"]["sourceIPAddress"]
 
 
+class S3EventNotificationEventBridgeBucket(DictWrapper):
+    @property
+    def name(self) -> str:
+        return self["name"]
+
+
+class S3EventBridgeNotificationObject(DictWrapper):
+    @property
+    def key(self) -> str:
+        """Object key"""
+        return unquote_plus(self["key"])
+
+    @property
+    def size(self) -> str:
+        """Object size"""
+        return self["size"]
+
+    @property
+    def etag(self) -> str:
+        """Object etag"""
+        return self["etag"]
+
+    @property
+    def version_id(self) -> str:
+        """Object version ID"""
+        return self["version-id"]
+
+    @property
+    def sequencer(self) -> str:
+        """Object key"""
+        return self["sequencer"]
+
+
+class S3EventBridgeNotificationDetail(DictWrapper):
+    @property
+    def version(self) -> str:
+        """Get the detail version"""
+        return self["version"]
+
+    @property
+    def bucket(self) -> S3EventNotificationEventBridgeBucket:
+        """Get the bucket name for the S3 notification"""
+        return S3EventNotificationEventBridgeBucket(self["bucket"])
+
+    @property
+    def object(self) -> S3EventBridgeNotificationObject:  # noqa: A003 # ignore shadowing built-in grammar
+        """Get the request-id for the S3 notification"""
+        return S3EventBridgeNotificationObject(self["object"])
+
+    @property
+    def request_id(self) -> str:
+        """Get the request-id for the S3 notification"""
+        return self["request-id"]
+
+    @property
+    def requester(self) -> str:
+        """Get the AWS account ID or AWS service principal of requester for the S3 notification"""
+        return self["requester"]
+
+    @property
+    def source_ip_address(self) -> Optional[str]:
+        """Get the source IP address of S3 request. Only present for events triggered by an S3 request."""
+        return self.get("source-ip-address")
+
+    @property
+    def reason(self) -> Optional[str]:
+        """Get the reason for the S3 notification.
+
+        For 'Object Created events', the S3 API used to create the object: `PutObject`, `POST Object`, `CopyObject`, or
+        `CompleteMultipartUpload`. For 'Object Deleted' events, this is set to `DeleteObject` when an object is deleted
+        by an S3 API call, or 'Lifecycle Expiration' when an object is deleted by an S3 Lifecycle expiration rule.
+        """
+        return self.get("reason")
+
+    @property
+    def deletion_type(self) -> Optional[str]:
+        """Get the deletion type for the S3 object in this notification.
+
+        For 'Object Deleted' events, when an unversioned object is deleted, or a versioned object is permanently deleted
+        this is set to 'Permanently Deleted'. When a delete marker is created for a versioned object, this is set to
+        'Delete Marker Created'.
+        """
+        return self.get("deletion-type")
+
+    @property
+    def restore_expiry_time(self) -> Optional[str]:
+        """Get the restore expiry time for the S3 object in this notification.
+
+        For 'Object Restore Completed' events, the time when the temporary copy of the object will be deleted from S3.
+        """
+        return self.get("restore-expiry-time")
+
+    @property
+    def source_storage_class(self) -> Optional[str]:
+        """Get the source storage class of the S3 object in this notification.
+
+        For 'Object Restore Initiated' and 'Object Restore Completed' events, the storage class of the object being
+        restored.
+        """
+        return self.get("source-storage-class")
+
+    @property
+    def destination_storage_class(self) -> Optional[str]:
+        """Get the destination storage class of the S3 object in this notification.
+
+        For 'Object Storage Class Changed' events, the new storage class of the object.
+        """
+        return self.get("destination-storage-class")
+
+    @property
+    def destination_access_tier(self) -> Optional[str]:
+        """Get the destination access tier of the S3 object in this notification.
+
+        For 'Object Access Tier Changed' events, the new access tier of the object.
+        """
+        return self.get("destination-access-tier")
+
+
+class S3EventBridgeNotificationEvent(EventBridgeEvent):
+    """Amazon S3EventBridge Event
+
+    Documentation:
+    --------------
+    - https://docs.aws.amazon.com/AmazonS3/latest/userguide/ev-events.html
+    """
+
+    @property
+    def detail(self) -> S3EventBridgeNotificationDetail:  # type: ignore[override]
+        """S3 notification details"""
+        return S3EventBridgeNotificationDetail(self["detail"])
+
+
 class S3Bucket(DictWrapper):
     @property
     def name(self) -> str:

docs/utilities/data_classes.md

@@ -82,6 +82,7 @@ Same example as above, but using the `event_source` decorator
 | [Rabbit MQ](#rabbit-mq)                                                   | `RabbitMQEvent`                                    |
 | [S3](#s3)                                                                 | `S3Event`                                          |
 | [S3 Object Lambda](#s3-object-lambda)                                     | `S3ObjectLambdaEvent`                              |
+| [S3 EventBridge Notification](#s3-eventbridge-notification)               | `S3EventBridgeNotificationEvent`                   |
 | [SES](#ses)                                                               | `SESEvent`                                         |
 | [SNS](#sns)                                                               | `SNSEvent`                                         |
 | [SQS](#sqs)                                                               | `SQSEvent`                                         |
@@ -1043,6 +1044,19 @@ This example is based on the AWS Blog post [Introducing Amazon S3 Object Lambda
         return {"status_code": 200}
     ```
 
+### S3 EventBridge Notification
+
+=== "app.py"
+
+    ```python
+    from aws_lambda_powertools.utilities.data_classes import event_source, S3EventBridgeNotificationEvent
+
+    @event_source(data_class=S3EventBridgeNotificationEvent)
+    def lambda_handler(event: S3EventBridgeNotificationEvent, context):
+        bucket_name = event.detail.bucket.name
+        file_key = event.detail.object.key
+    ```
+
 ### SES
 
 === "app.py"