Module aws_lambda_powertools.utilities.data_classes.s3_event

Expand source code 
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.event_bridge_event import (
    EventBridgeEvent,
)


class S3Identity(DictWrapper):
    @property
    def principal_id(self) -> str:
        return self["principalId"]


class S3RequestParameters(DictWrapper):
    @property
    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:
        return self["s3"]["bucket"]["name"]

    @property
    def owner_identity(self) -> S3Identity:
        return S3Identity(self["s3"]["bucket"]["ownerIdentity"])

    @property
    def arn(self) -> str:
        return self["s3"]["bucket"]["arn"]


class S3Object(DictWrapper):
    @property
    def key(self) -> str:
        """Object key"""
        return self["s3"]["object"]["key"]

    @property
    def size(self) -> int:
        """Object byte size"""
        return int(self["s3"]["object"]["size"])

    @property
    def etag(self) -> str:
        """object eTag"""
        return self["s3"]["object"]["eTag"]

    @property
    def version_id(self) -> Optional[str]:
        """Object version if bucket is versioning-enabled, otherwise null"""
        return self["s3"]["object"].get("versionId")

    @property
    def sequencer(self) -> str:
        """A string representation of a hexadecimal value used to determine event sequence,
        only used with PUTs and DELETEs
        """
        return self["s3"]["object"]["sequencer"]


class S3Message(DictWrapper):
    @property
    def s3_schema_version(self) -> str:
        return self["s3"]["s3SchemaVersion"]

    @property
    def configuration_id(self) -> str:
        """ID found in the bucket notification configuration"""
        return self["s3"]["configurationId"]

    @property
    def bucket(self) -> S3Bucket:
        return S3Bucket(self._data)

    @property
    def get_object(self) -> S3Object:
        """Get the `object` property as an S3Object"""
        # Note: this name conflicts with existing python builtins
        return S3Object(self._data)


class S3EventRecordGlacierRestoreEventData(DictWrapper):
    @property
    def lifecycle_restoration_expiry_time(self) -> str:
        """Time when the object restoration will be expired."""
        return self["restoreEventData"]["lifecycleRestorationExpiryTime"]

    @property
    def lifecycle_restore_storage_class(self) -> str:
        """Source storage class for restore"""
        return self["restoreEventData"]["lifecycleRestoreStorageClass"]


class S3EventRecordGlacierEventData(DictWrapper):
    @property
    def restore_event_data(self) -> S3EventRecordGlacierRestoreEventData:
        """The restoreEventData key contains attributes related to your restore request.

        The glacierEventData key is only visible for s3:ObjectRestore:Completed events
        """
        return S3EventRecordGlacierRestoreEventData(self._data)


class S3EventRecord(DictWrapper):
    @property
    def event_version(self) -> str:
        """The eventVersion key value contains a major and minor version in the form <major>.<minor>."""
        return self["eventVersion"]

    @property
    def event_source(self) -> str:
        """The AWS service from which the S3 event originated. For S3, this is aws:s3"""
        return self["eventSource"]

    @property
    def aws_region(self) -> str:
        """aws region eg: us-east-1"""
        return self["awsRegion"]

    @property
    def event_time(self) -> str:
        """The time, in ISO-8601 format, for example, 1970-01-01T00:00:00.000Z, when S3 finished
        processing the request"""
        return self["eventTime"]

    @property
    def event_name(self) -> str:
        """Event type"""
        return self["eventName"]

    @property
    def user_identity(self) -> S3Identity:
        return S3Identity(self["userIdentity"])

    @property
    def request_parameters(self) -> S3RequestParameters:
        return S3RequestParameters(self._data)

    @property
    def response_elements(self) -> Dict[str, str]:
        """The responseElements key value is useful if you want to trace a request by following up with AWS Support.

        Both x-amz-request-id and x-amz-id-2 help Amazon S3 trace an individual request. These values are the same
        as those that Amazon S3 returns in the response to the request that initiates the events, so they can be
        used to match the event to the request.
        """
        return self["responseElements"]

    @property
    def s3(self) -> S3Message:
        return S3Message(self._data)

    @property
    def glacier_event_data(self) -> Optional[S3EventRecordGlacierEventData]:
        """The glacierEventData key is only visible for s3:ObjectRestore:Completed events."""
        item = self.get("glacierEventData")
        return None if item is None else S3EventRecordGlacierEventData(item)


class S3Event(DictWrapper):
    """S3 event notification

    Documentation:
    -------------
    - https://docs.aws.amazon.com/lambda/latest/dg/with-s3.html
    - https://docs.aws.amazon.com/AmazonS3/latest/dev/NotificationHowTo.html
    - https://docs.aws.amazon.com/AmazonS3/latest/dev/notification-content-structure.html
    """

    @property
    def records(self) -> Iterator[S3EventRecord]:
        for record in self["Records"]:
            yield S3EventRecord(record)

    @property
    def record(self) -> S3EventRecord:
        """Get the first s3 event record"""
        return next(self.records)

    @property
    def bucket_name(self) -> str:
        """Get the bucket name for the first s3 event record"""
        return self["Records"][0]["s3"]["bucket"]["name"]

    @property
    def object_key(self) -> str:
        """Get the object key for the first s3 event record and unquote plus"""
        return unquote_plus(self["Records"][0]["s3"]["object"]["key"])

Classes

class S3Bucket (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

Provides a single read only access to a wrapper dict

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
Expand source code 
class S3Bucket(DictWrapper):
    @property
    def name(self) -> str:
        return self["s3"]["bucket"]["name"]

    @property
    def owner_identity(self) -> S3Identity:
        return S3Identity(self["s3"]["bucket"]["ownerIdentity"])

    @property
    def arn(self) -> str:
        return self["s3"]["bucket"]["arn"]

Ancestors

  • DictWrapper
  • collections.abc.Mapping
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container

Instance variables

var arn : str
Expand source code 
@property
def arn(self) -> str:
    return self["s3"]["bucket"]["arn"]
var name : str
Expand source code 
@property
def name(self) -> str:
    return self["s3"]["bucket"]["name"]
var owner_identity : S3Identity
Expand source code 
@property
def owner_identity(self) -> S3Identity:
    return S3Identity(self["s3"]["bucket"]["ownerIdentity"])

Inherited members

class S3Event (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

S3 event notification

Documentation:

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
Expand source code 
class S3Event(DictWrapper):
    """S3 event notification

    Documentation:
    -------------
    - https://docs.aws.amazon.com/lambda/latest/dg/with-s3.html
    - https://docs.aws.amazon.com/AmazonS3/latest/dev/NotificationHowTo.html
    - https://docs.aws.amazon.com/AmazonS3/latest/dev/notification-content-structure.html
    """

    @property
    def records(self) -> Iterator[S3EventRecord]:
        for record in self["Records"]:
            yield S3EventRecord(record)

    @property
    def record(self) -> S3EventRecord:
        """Get the first s3 event record"""
        return next(self.records)

    @property
    def bucket_name(self) -> str:
        """Get the bucket name for the first s3 event record"""
        return self["Records"][0]["s3"]["bucket"]["name"]

    @property
    def object_key(self) -> str:
        """Get the object key for the first s3 event record and unquote plus"""
        return unquote_plus(self["Records"][0]["s3"]["object"]["key"])

Ancestors

  • DictWrapper
  • collections.abc.Mapping
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container

Instance variables

var bucket_name : str

Get the bucket name for the first s3 event record

Expand source code 
@property
def bucket_name(self) -> str:
    """Get the bucket name for the first s3 event record"""
    return self["Records"][0]["s3"]["bucket"]["name"]
var object_key : str

Get the object key for the first s3 event record and unquote plus

Expand source code 
@property
def object_key(self) -> str:
    """Get the object key for the first s3 event record and unquote plus"""
    return unquote_plus(self["Records"][0]["s3"]["object"]["key"])
var record : S3EventRecord

Get the first s3 event record

Expand source code 
@property
def record(self) -> S3EventRecord:
    """Get the first s3 event record"""
    return next(self.records)
var records : Iterator[S3EventRecord]
Expand source code 
@property
def records(self) -> Iterator[S3EventRecord]:
    for record in self["Records"]:
        yield S3EventRecord(record)

Inherited members

class S3EventBridgeNotificationDetail (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

Provides a single read only access to a wrapper dict

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
Expand source code 
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")

Ancestors

  • DictWrapper
  • collections.abc.Mapping
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container

Instance variables

var bucket : S3EventNotificationEventBridgeBucket

Get the bucket name for the S3 notification

Expand source code 
@property
def bucket(self) -> S3EventNotificationEventBridgeBucket:
    """Get the bucket name for the S3 notification"""
    return S3EventNotificationEventBridgeBucket(self["bucket"])
var deletion_type : 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'.

Expand source code 
@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")
var destination_access_tier : 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.

Expand source code 
@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")
var destination_storage_class : 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.

Expand source code 
@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")
var object : S3EventBridgeNotificationObject

Get the request-id for the S3 notification

Expand source code 
@property
def object(self) -> S3EventBridgeNotificationObject:  # noqa: A003 # ignore shadowing built-in grammar
    """Get the request-id for the S3 notification"""
    return S3EventBridgeNotificationObject(self["object"])
var reason : 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.

Expand source code 
@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")
var request_id : str

Get the request-id for the S3 notification

Expand source code 
@property
def request_id(self) -> str:
    """Get the request-id for the S3 notification"""
    return self["request-id"]
var requester : str

Get the AWS account ID or AWS service principal of requester for the S3 notification

Expand source code 
@property
def requester(self) -> str:
    """Get the AWS account ID or AWS service principal of requester for the S3 notification"""
    return self["requester"]
var restore_expiry_time : 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.

Expand source code 
@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")
var source_ip_address : Optional[str]

Get the source IP address of S3 request. Only present for events triggered by an S3 request.

Expand source code 
@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")
var source_storage_class : 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.

Expand source code 
@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")
var version : str

Get the detail version

Expand source code 
@property
def version(self) -> str:
    """Get the detail version"""
    return self["version"]

Inherited members

class S3EventBridgeNotificationEvent (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

Amazon S3EventBridge Event

Documentation:

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
Expand source code 
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"])

Ancestors

  • EventBridgeEvent
  • DictWrapper
  • collections.abc.Mapping
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container

Instance variables

var detail : S3EventBridgeNotificationDetail

S3 notification details

Expand source code 
@property
def detail(self) -> S3EventBridgeNotificationDetail:  # type: ignore[override]
    """S3 notification details"""
    return S3EventBridgeNotificationDetail(self["detail"])

Inherited members

class S3EventBridgeNotificationObject (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

Provides a single read only access to a wrapper dict

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
Expand source code 
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"]

Ancestors

  • DictWrapper
  • collections.abc.Mapping
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container

Instance variables

var etag : str

Object etag

Expand source code 
@property
def etag(self) -> str:
    """Object etag"""
    return self["etag"]
var key : str

Object key

Expand source code 
@property
def key(self) -> str:
    """Object key"""
    return unquote_plus(self["key"])
var sequencer : str

Object key

Expand source code 
@property
def sequencer(self) -> str:
    """Object key"""
    return self["sequencer"]
var size : str

Object size

Expand source code 
@property
def size(self) -> str:
    """Object size"""
    return self["size"]
var version_id : str

Object version ID

Expand source code 
@property
def version_id(self) -> str:
    """Object version ID"""
    return self["version-id"]

Inherited members

class S3EventNotificationEventBridgeBucket (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

Provides a single read only access to a wrapper dict

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
Expand source code 
class S3EventNotificationEventBridgeBucket(DictWrapper):
    @property
    def name(self) -> str:
        return self["name"]

Ancestors

  • DictWrapper
  • collections.abc.Mapping
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container

Instance variables

var name : str
Expand source code 
@property
def name(self) -> str:
    return self["name"]

Inherited members

class S3EventRecord (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

Provides a single read only access to a wrapper dict

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
Expand source code 
class S3EventRecord(DictWrapper):
    @property
    def event_version(self) -> str:
        """The eventVersion key value contains a major and minor version in the form <major>.<minor>."""
        return self["eventVersion"]

    @property
    def event_source(self) -> str:
        """The AWS service from which the S3 event originated. For S3, this is aws:s3"""
        return self["eventSource"]

    @property
    def aws_region(self) -> str:
        """aws region eg: us-east-1"""
        return self["awsRegion"]

    @property
    def event_time(self) -> str:
        """The time, in ISO-8601 format, for example, 1970-01-01T00:00:00.000Z, when S3 finished
        processing the request"""
        return self["eventTime"]

    @property
    def event_name(self) -> str:
        """Event type"""
        return self["eventName"]

    @property
    def user_identity(self) -> S3Identity:
        return S3Identity(self["userIdentity"])

    @property
    def request_parameters(self) -> S3RequestParameters:
        return S3RequestParameters(self._data)

    @property
    def response_elements(self) -> Dict[str, str]:
        """The responseElements key value is useful if you want to trace a request by following up with AWS Support.

        Both x-amz-request-id and x-amz-id-2 help Amazon S3 trace an individual request. These values are the same
        as those that Amazon S3 returns in the response to the request that initiates the events, so they can be
        used to match the event to the request.
        """
        return self["responseElements"]

    @property
    def s3(self) -> S3Message:
        return S3Message(self._data)

    @property
    def glacier_event_data(self) -> Optional[S3EventRecordGlacierEventData]:
        """The glacierEventData key is only visible for s3:ObjectRestore:Completed events."""
        item = self.get("glacierEventData")
        return None if item is None else S3EventRecordGlacierEventData(item)

Ancestors

  • DictWrapper
  • collections.abc.Mapping
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container

Instance variables

var aws_region : str

aws region eg: us-east-1

Expand source code 
@property
def aws_region(self) -> str:
    """aws region eg: us-east-1"""
    return self["awsRegion"]
var event_name : str

Event type

Expand source code 
@property
def event_name(self) -> str:
    """Event type"""
    return self["eventName"]
var event_source : str

The AWS service from which the S3 event originated. For S3, this is aws:s3

Expand source code 
@property
def event_source(self) -> str:
    """The AWS service from which the S3 event originated. For S3, this is aws:s3"""
    return self["eventSource"]
var event_time : str

The time, in ISO-8601 format, for example, 1970-01-01T00:00:00.000Z, when S3 finished processing the request

Expand source code 
@property
def event_time(self) -> str:
    """The time, in ISO-8601 format, for example, 1970-01-01T00:00:00.000Z, when S3 finished
    processing the request"""
    return self["eventTime"]
var event_version : str

The eventVersion key value contains a major and minor version in the form ..

Expand source code 
@property
def event_version(self) -> str:
    """The eventVersion key value contains a major and minor version in the form <major>.<minor>."""
    return self["eventVersion"]
var glacier_event_data : Optional[S3EventRecordGlacierEventData]

The glacierEventData key is only visible for s3:ObjectRestore:Completed events.

Expand source code 
@property
def glacier_event_data(self) -> Optional[S3EventRecordGlacierEventData]:
    """The glacierEventData key is only visible for s3:ObjectRestore:Completed events."""
    item = self.get("glacierEventData")
    return None if item is None else S3EventRecordGlacierEventData(item)
var request_parameters : S3RequestParameters
Expand source code 
@property
def request_parameters(self) -> S3RequestParameters:
    return S3RequestParameters(self._data)
var response_elements : Dict[str, str]

The responseElements key value is useful if you want to trace a request by following up with AWS Support.

Both x-amz-request-id and x-amz-id-2 help Amazon S3 trace an individual request. These values are the same as those that Amazon S3 returns in the response to the request that initiates the events, so they can be used to match the event to the request.

Expand source code 
@property
def response_elements(self) -> Dict[str, str]:
    """The responseElements key value is useful if you want to trace a request by following up with AWS Support.

    Both x-amz-request-id and x-amz-id-2 help Amazon S3 trace an individual request. These values are the same
    as those that Amazon S3 returns in the response to the request that initiates the events, so they can be
    used to match the event to the request.
    """
    return self["responseElements"]
var s3 : S3Message
Expand source code 
@property
def s3(self) -> S3Message:
    return S3Message(self._data)
var user_identity : S3Identity
Expand source code 
@property
def user_identity(self) -> S3Identity:
    return S3Identity(self["userIdentity"])

Inherited members

class S3EventRecordGlacierEventData (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

Provides a single read only access to a wrapper dict

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
Expand source code 
class S3EventRecordGlacierEventData(DictWrapper):
    @property
    def restore_event_data(self) -> S3EventRecordGlacierRestoreEventData:
        """The restoreEventData key contains attributes related to your restore request.

        The glacierEventData key is only visible for s3:ObjectRestore:Completed events
        """
        return S3EventRecordGlacierRestoreEventData(self._data)

Ancestors

  • DictWrapper
  • collections.abc.Mapping
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container

Instance variables

var restore_event_data : S3EventRecordGlacierRestoreEventData

The restoreEventData key contains attributes related to your restore request.

The glacierEventData key is only visible for s3:ObjectRestore:Completed events

Expand source code 
@property
def restore_event_data(self) -> S3EventRecordGlacierRestoreEventData:
    """The restoreEventData key contains attributes related to your restore request.

    The glacierEventData key is only visible for s3:ObjectRestore:Completed events
    """
    return S3EventRecordGlacierRestoreEventData(self._data)

Inherited members

class S3EventRecordGlacierRestoreEventData (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

Provides a single read only access to a wrapper dict

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
Expand source code 
class S3EventRecordGlacierRestoreEventData(DictWrapper):
    @property
    def lifecycle_restoration_expiry_time(self) -> str:
        """Time when the object restoration will be expired."""
        return self["restoreEventData"]["lifecycleRestorationExpiryTime"]

    @property
    def lifecycle_restore_storage_class(self) -> str:
        """Source storage class for restore"""
        return self["restoreEventData"]["lifecycleRestoreStorageClass"]

Ancestors

  • DictWrapper
  • collections.abc.Mapping
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container

Instance variables

var lifecycle_restoration_expiry_time : str

Time when the object restoration will be expired.

Expand source code 
@property
def lifecycle_restoration_expiry_time(self) -> str:
    """Time when the object restoration will be expired."""
    return self["restoreEventData"]["lifecycleRestorationExpiryTime"]
var lifecycle_restore_storage_class : str

Source storage class for restore

Expand source code 
@property
def lifecycle_restore_storage_class(self) -> str:
    """Source storage class for restore"""
    return self["restoreEventData"]["lifecycleRestoreStorageClass"]

Inherited members

class S3Identity (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

Provides a single read only access to a wrapper dict

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
Expand source code 
class S3Identity(DictWrapper):
    @property
    def principal_id(self) -> str:
        return self["principalId"]

Ancestors

  • DictWrapper
  • collections.abc.Mapping
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container

Instance variables

var principal_id : str
Expand source code 
@property
def principal_id(self) -> str:
    return self["principalId"]

Inherited members

class S3Message (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

Provides a single read only access to a wrapper dict

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
Expand source code 
class S3Message(DictWrapper):
    @property
    def s3_schema_version(self) -> str:
        return self["s3"]["s3SchemaVersion"]

    @property
    def configuration_id(self) -> str:
        """ID found in the bucket notification configuration"""
        return self["s3"]["configurationId"]

    @property
    def bucket(self) -> S3Bucket:
        return S3Bucket(self._data)

    @property
    def get_object(self) -> S3Object:
        """Get the `object` property as an S3Object"""
        # Note: this name conflicts with existing python builtins
        return S3Object(self._data)

Ancestors

  • DictWrapper
  • collections.abc.Mapping
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container

Instance variables

var bucket : S3Bucket
Expand source code 
@property
def bucket(self) -> S3Bucket:
    return S3Bucket(self._data)
var configuration_id : str

ID found in the bucket notification configuration

Expand source code 
@property
def configuration_id(self) -> str:
    """ID found in the bucket notification configuration"""
    return self["s3"]["configurationId"]
var get_object : S3Object

Get the object property as an S3Object

Expand source code 
@property
def get_object(self) -> S3Object:
    """Get the `object` property as an S3Object"""
    # Note: this name conflicts with existing python builtins
    return S3Object(self._data)
var s3_schema_version : str
Expand source code 
@property
def s3_schema_version(self) -> str:
    return self["s3"]["s3SchemaVersion"]

Inherited members

class S3Object (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

Provides a single read only access to a wrapper dict

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
Expand source code 
class S3Object(DictWrapper):
    @property
    def key(self) -> str:
        """Object key"""
        return self["s3"]["object"]["key"]

    @property
    def size(self) -> int:
        """Object byte size"""
        return int(self["s3"]["object"]["size"])

    @property
    def etag(self) -> str:
        """object eTag"""
        return self["s3"]["object"]["eTag"]

    @property
    def version_id(self) -> Optional[str]:
        """Object version if bucket is versioning-enabled, otherwise null"""
        return self["s3"]["object"].get("versionId")

    @property
    def sequencer(self) -> str:
        """A string representation of a hexadecimal value used to determine event sequence,
        only used with PUTs and DELETEs
        """
        return self["s3"]["object"]["sequencer"]

Ancestors

  • DictWrapper
  • collections.abc.Mapping
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container

Instance variables

var etag : str

object eTag

Expand source code 
@property
def etag(self) -> str:
    """object eTag"""
    return self["s3"]["object"]["eTag"]
var key : str

Object key

Expand source code 
@property
def key(self) -> str:
    """Object key"""
    return self["s3"]["object"]["key"]
var sequencer : str

A string representation of a hexadecimal value used to determine event sequence, only used with PUTs and DELETEs

Expand source code 
@property
def sequencer(self) -> str:
    """A string representation of a hexadecimal value used to determine event sequence,
    only used with PUTs and DELETEs
    """
    return self["s3"]["object"]["sequencer"]
var size : int

Object byte size

Expand source code 
@property
def size(self) -> int:
    """Object byte size"""
    return int(self["s3"]["object"]["size"])
var version_id : Optional[str]

Object version if bucket is versioning-enabled, otherwise null

Expand source code 
@property
def version_id(self) -> Optional[str]:
    """Object version if bucket is versioning-enabled, otherwise null"""
    return self["s3"]["object"].get("versionId")

Inherited members

class S3RequestParameters (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

Provides a single read only access to a wrapper dict

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
Expand source code 
class S3RequestParameters(DictWrapper):
    @property
    def source_ip_address(self) -> str:
        return self["requestParameters"]["sourceIPAddress"]

Ancestors

  • DictWrapper
  • collections.abc.Mapping
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container

Instance variables

var source_ip_address : str
Expand source code 
@property
def source_ip_address(self) -> str:
    return self["requestParameters"]["sourceIPAddress"]

Inherited members