Module aws_lambda_powertools.utilities.data_classes.s3_event

Classes

class S3Bucket (data: Dict[str, Any])

Provides a single read only access to a wrapper dict

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

• DictWrapper:
  • get
  • raw_event

class S3Event (data: Dict[str, Any])

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

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

• DictWrapper:
  • get
  • raw_event

class S3EventRecord (data: Dict[str, Any])

Provides a single read only access to a wrapper dict

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

• DictWrapper:
  • get
  • raw_event

class S3EventRecordGlacierEventData (data: Dict[str, Any])

Provides a single read only access to a wrapper dict

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

• DictWrapper:
  • get
  • raw_event

class S3EventRecordGlacierRestoreEventData (data: Dict[str, Any])

Provides a single read only access to a wrapper dict

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

• DictWrapper:
  • get
  • raw_event

class S3Identity (data: Dict[str, Any])

Provides a single read only access to a wrapper dict

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

• DictWrapper:
  • get
  • raw_event

class S3Message (data: Dict[str, Any])

Provides a single read only access to a wrapper dict

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

• DictWrapper:
  • get
  • raw_event

class S3Object (data: Dict[str, Any])

Provides a single read only access to a wrapper dict

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

• DictWrapper:
  • get
  • raw_event

class S3RequestParameters (data: Dict[str, Any])

Provides a single read only access to a wrapper dict

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

• DictWrapper:
  • get
  • raw_event