feat(data-classes): adds support for S3 event notifications through EventBridge

Author: ivica-k

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

tests/functional/data_classes/test_s3_eventbridge_notification.py

@@ -0,0 +1,126 @@
+from aws_lambda_powertools.utilities.data_classes.s3_event import (
+    S3EventBridgeNotificationDetail,
+    S3EventBridgeNotificationEvent,
+    S3EventBridgeNotificationObject,
+)
+from tests.functional.utils import load_event
+
+
+def test_s3_eventbridge_notification_detail_parsed_object_created():
+    event = S3EventBridgeNotificationEvent(load_event("s3EventBridgeNotificationObjectCreatedEvent.json"))
+    bucket_name = "example-bucket"
+    deletion_type = None
+    destination_access_tier = None
+    destination_storage_class = None
+    detail: S3EventBridgeNotificationDetail = event.detail
+    _object: S3EventBridgeNotificationObject = detail.object
+    reason = "PutObject"
+    request_id = "57H08PA84AB1JZW0"
+    requester = "123456789012"
+    restore_expiry_time = None
+    source_ip_address = "34.252.34.74"
+    source_storage_class = None
+    version = "0"
+
+    assert bucket_name == event.detail.bucket.name
+    assert deletion_type == event.detail.deletion_type
+    assert destination_access_tier == event.detail.destination_access_tier
+    assert destination_storage_class == event.detail.destination_storage_class
+    assert _object == event.detail.object
+    assert reason == event.detail.reason
+    assert request_id == event.detail.request_id
+    assert requester == event.detail.requester
+    assert restore_expiry_time == event.detail.restore_expiry_time
+    assert source_ip_address == event.detail.source_ip_address
+    assert source_storage_class == event.detail.source_storage_class
+    assert version == event.version
+
+
+def test_s3_eventbridge_notification_detail_parsed_object_deleted():
+    event = S3EventBridgeNotificationEvent(load_event("s3EventBridgeNotificationObjectDeletedEvent.json"))
+    bucket_name = "example-bucket"
+    deletion_type = "Delete Marker Created"
+    destination_access_tier = None
+    destination_storage_class = None
+    detail: S3EventBridgeNotificationDetail = event.detail
+    _object: S3EventBridgeNotificationObject = detail.object
+    reason = "DeleteObject"
+    request_id = "0BH729840619AG5K"
+    requester = "123456789012"
+    restore_expiry_time = None
+    source_ip_address = "34.252.34.74"
+    source_storage_class = None
+    version = "0"
+
+    assert bucket_name == event.detail.bucket.name
+    assert deletion_type == event.detail.deletion_type
+    assert destination_access_tier == event.detail.destination_access_tier
+    assert destination_storage_class == event.detail.destination_storage_class
+    assert _object == event.detail.object
+    assert reason == event.detail.reason
+    assert request_id == event.detail.request_id
+    assert requester == event.detail.requester
+    assert restore_expiry_time == event.detail.restore_expiry_time
+    assert source_ip_address == event.detail.source_ip_address
+    assert source_storage_class == event.detail.source_storage_class
+    assert version == event.version
+
+
+def test_s3_eventbridge_notification_detail_parsed_object_expired():
+    event = S3EventBridgeNotificationEvent(load_event("s3EventBridgeNotificationObjectExpiredEvent.json"))
+    bucket_name = "example-bucket"
+    deletion_type = "Delete Marker Created"
+    destination_access_tier = None
+    destination_storage_class = None
+    detail: S3EventBridgeNotificationDetail = event.detail
+    _object: S3EventBridgeNotificationObject = detail.object
+    reason = "Lifecycle Expiration"
+    request_id = "20EB74C14654DC47"
+    requester = "s3.amazonaws.com"
+    restore_expiry_time = None
+    source_ip_address = None
+    source_storage_class = None
+    version = "0"
+
+    assert bucket_name == event.detail.bucket.name
+    assert deletion_type == event.detail.deletion_type
+    assert destination_access_tier == event.detail.destination_access_tier
+    assert destination_storage_class == event.detail.destination_storage_class
+    assert _object == event.detail.object
+    assert reason == event.detail.reason
+    assert request_id == event.detail.request_id
+    assert requester == event.detail.requester
+    assert restore_expiry_time == event.detail.restore_expiry_time
+    assert source_ip_address == event.detail.source_ip_address
+    assert source_storage_class == event.detail.source_storage_class
+    assert version == event.version
+
+
+def test_s3_eventbridge_notification_detail_parsed_object_restore_completed():
+    event = S3EventBridgeNotificationEvent(load_event("s3EventBridgeNotificationObjectRestoreCompletedEvent.json"))
+    bucket_name = "example-bucket"
+    deletion_type = None
+    destination_access_tier = None
+    destination_storage_class = None
+    detail: S3EventBridgeNotificationDetail = event.detail
+    _object: S3EventBridgeNotificationObject = detail.object
+    reason = None
+    request_id = "189F19CB7FB1B6A4"
+    requester = "s3.amazonaws.com"
+    restore_expiry_time = "2021-11-13T00:00:00Z"
+    source_ip_address = None
+    source_storage_class = "GLACIER"
+    version = "0"
+
+    assert bucket_name == event.detail.bucket.name
+    assert deletion_type == event.detail.deletion_type
+    assert destination_access_tier == event.detail.destination_access_tier
+    assert destination_storage_class == event.detail.destination_storage_class
+    assert _object == event.detail.object
+    assert reason == event.detail.reason
+    assert request_id == event.detail.request_id
+    assert requester == event.detail.requester
+    assert restore_expiry_time == event.detail.restore_expiry_time
+    assert source_ip_address == event.detail.source_ip_address
+    assert source_storage_class == event.detail.source_storage_class
+    assert version == event.version