feat(parser): adds support for S3 event notifications through EventBridge

Author: ivica-k

aws_lambda_powertools/utilities/data_classes/s3_event.py

@@ -29,27 +29,27 @@ class S3EventBridgeNotificationObject(DictWrapper):
     @property
     def key(self) -> str:
         """Object key"""
-        return unquote_plus(self["object"]["key"])
+        return unquote_plus(self["key"])
 
     @property
     def size(self) -> str:
         """Object size"""
-        return self["object"]["size"]
+        return self["size"]
 
     @property
     def etag(self) -> str:
         """Object etag"""
-        return self["object"]["etag"]
+        return self["etag"]
 
     @property
     def version_id(self) -> str:
         """Object version ID"""
-        return self["object"]["version-id"]
+        return self["version-id"]
 
     @property
     def sequencer(self) -> str:
         """Object key"""
-        return self["object"]["sequencer"]
+        return self["sequencer"]
 
 
 class S3EventBridgeNotificationDetail(DictWrapper):
@@ -61,27 +61,27 @@ def version(self) -> str:
     @property
     def bucket(self) -> S3EventNotificationEventBridgeBucket:
         """Get the bucket name for the S3 notification"""
-        return S3EventNotificationEventBridgeBucket(self["detail"]["bucket"])
+        return S3EventNotificationEventBridgeBucket(self["bucket"])
 
     @property
     def object(self) -> S3EventBridgeNotificationObject:  # noqa: A003
         """Get the request-id for the S3 notification"""
-        return S3EventBridgeNotificationObject(self["detail"]["object"])
+        return S3EventBridgeNotificationObject(self["object"])
 
     @property
     def request_id(self) -> str:
         """Get the request-id for the S3 notification"""
-        return self["detail"]["request-id"]
+        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["detail"]["requester"]
+        return self["requester"]
 
     @property
     def source_ip_address(self) -> str:
         """Get the source IP address of S3 request. Only present for events triggered by an S3 request."""
-        return self["detail"]["source-ip-address"]
+        return self["source-ip-address"]
 
     @property
     def reason(self) -> Optional[str]:
@@ -91,7 +91,7 @@ def reason(self) -> Optional[str]:
         `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["detail"].get("reason")
+        return self.get("reason")
 
     @property
     def deletion_type(self) -> Optional[str]:
@@ -101,7 +101,7 @@ def deletion_type(self) -> Optional[str]:
         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["detail"].get("deletion-type")
+        return self.get("deletion-type")
 
     @property
     def restore_expiry_time(self) -> Optional[str]: