diff --git a/aws_lambda_powertools/utilities/data_classes/s3_event.py b/aws_lambda_powertools/utilities/data_classes/s3_event.py index bf404f1ecbf..b601ffc9e68 100644 --- a/aws_lambda_powertools/utilities/data_classes/s3_event.py +++ b/aws_lambda_powertools/utilities/data_classes/s3_event.py @@ -215,9 +215,14 @@ def bucket(self) -> S3Bucket: @property def get_object(self) -> S3Object: - """Get the `object` property as an S3Object""" - # Note: this name conflicts with existing python builtins - return S3Object(self["object"]) + """Get the `object` property as an S3Object + + Note: IntelligentTiering events use 'get_object' as the actual key name, + while other S3 events use 'object'. This method handles both cases. + """ + # IntelligentTiering events use 'get_object', others use 'object' + object_data = self.get("get_object") or self["object"] + return S3Object(object_data) class S3EventRecordGlacierRestoreEventData(DictWrapper): @@ -242,6 +247,16 @@ def restore_event_data(self) -> S3EventRecordGlacierRestoreEventData: return S3EventRecordGlacierRestoreEventData(self["restoreEventData"]) +class S3EventRecordIntelligentTieringEventData(DictWrapper): + @property + def destination_access_tier(self) -> str: + """The new access tier for the object. + + The intelligentTieringEventData key is only visible for IntelligentTiering events. + """ + return self["destinationAccessTier"] + + class S3EventRecord(DictWrapper): @property def event_version(self) -> str: @@ -297,6 +312,12 @@ def glacier_event_data(self) -> S3EventRecordGlacierEventData | None: item = self.get("glacierEventData") return None if item is None else S3EventRecordGlacierEventData(item) + @property + def intelligent_tiering_event_data(self) -> S3EventRecordIntelligentTieringEventData | None: + """The intelligentTieringEventData key is only visible for IntelligentTiering events.""" + item = self.get("intelligentTieringEventData") + return None if item is None else S3EventRecordIntelligentTieringEventData(item) + class S3Event(DictWrapper): """S3 event notification @@ -325,5 +346,12 @@ def bucket_name(self) -> str: @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"]) + """Get the object key for the first s3 event record and unquote plus + + Note: IntelligentTiering events use 'get_object' as the key name, + while other S3 events use 'object'. This method handles both cases. + """ + s3_data = self["Records"][0]["s3"] + # IntelligentTiering events use 'get_object', others use 'object' + object_data = s3_data.get("get_object") or s3_data["object"] + return unquote_plus(object_data["key"]) diff --git a/aws_lambda_powertools/utilities/parser/models/s3.py b/aws_lambda_powertools/utilities/parser/models/s3.py index 11369721e8e..d53a0fe5655 100644 --- a/aws_lambda_powertools/utilities/parser/models/s3.py +++ b/aws_lambda_powertools/utilities/parser/models/s3.py @@ -40,6 +40,16 @@ class S3EventRecordGlacierEventData(BaseModel): ) +class S3EventRecordIntelligentTieringEventData(BaseModel): + destinationAccessTier: str = Field( + description="The new access tier for the object. For IntelligentTiering events.", + examples=[ + "ARCHIVE_ACCESS", + "DEEP_ARCHIVE_ACCESS", + ], + ) + + class S3Identity(BaseModel): principalId: str = Field( description="Amazon identifier of the user, role, account or services who caused the event.", @@ -178,8 +188,9 @@ class S3Message(BaseModel): }, ], ) - object: S3Object = Field( - description="The S3 object object.", + object: Optional[S3Object] = Field( + default=None, + description="The S3 object object. Used by most S3 event types.", examples=[ { "key": "b21b84d653bb07b05b1e6b33684dc11b", @@ -189,6 +200,20 @@ class S3Message(BaseModel): }, ], ) # noqa: A003 + get_object: Optional[S3Object] = Field( + default=None, + alias="get_object", + description="The S3 object object. Used by IntelligentTiering events instead of 'object'.", + examples=[ + { + "key": "myobject", + "size": 252294, + "eTag": "4e9270240d7d62d5ee8dbfcb7a7a3279", + "versionId": "tiogA9Ga7Xi49yfJ6lkeTxPYx7ZK75yn", + "sequencer": "0066A8D0E77DE42BC5", + }, + ], + ) class S3EventNotificationObjectModel(BaseModel): @@ -449,11 +474,22 @@ class S3RecordModel(BaseModel): }, ], ) + intelligentTieringEventData: Optional[S3EventRecordIntelligentTieringEventData] = Field( + default=None, + description="The Intelligent-Tiering event data object.", + examples=[ + { + "destinationAccessTier": "ARCHIVE_ACCESS", + }, + ], + ) @model_validator(mode="before") def validate_s3_object(cls, values): event_name = values.get("eventName") - s3_object = values.get("s3").get("object") + s3_data = values.get("s3") + # IntelligentTiering events use 'get_object' instead of 'object' + s3_object = s3_data.get("object") or s3_data.get("get_object") if ":Delete" not in event_name and (s3_object.get("size") is None or s3_object.get("eTag") is None): raise ValueError( "Size and eTag fields are required for all events except ObjectRemoved:* and LifecycleExpiration:*.", diff --git a/tests/events/s3EventIntelligentTiering.json b/tests/events/s3EventIntelligentTiering.json new file mode 100644 index 00000000000..06c7ca469ac --- /dev/null +++ b/tests/events/s3EventIntelligentTiering.json @@ -0,0 +1,42 @@ +{ + "Records": [ + { + "eventVersion": "2.3", + "eventSource": "aws:s3", + "awsRegion": "ap-southeast-2", + "eventTime": "2025-09-29T00:47:23.967Z", + "eventName": "IntelligentTiering", + "userIdentity": { + "principalId": "s3.amazonaws.com" + }, + "requestParameters": { + "sourceIPAddress": "s3.amazonaws.com" + }, + "responseElements": { + "x-amz-request-id": "4C05BE601E268FCD", + "x-amz-id-2": "Wt0dgzBP/wQ47PWP5BqoD4nAZC08NOjfdZj2prx5gVPBCI+YQtrx/UKVlnT63Dj+vse4x5koWp61HaoVyz6fUFe1yLkhNpK3" + }, + "s3": { + "s3SchemaVersion": "1.0", + "configurationId": "ZDM3MGVjY2MtZTlmNC00ZTk0LThiNjItMWJiNTMwOTYwY2Rl", + "bucket": { + "name": "mybucket", + "ownerIdentity": { + "principalId": "ANY597T3BWFY2" + }, + "arn": "arn:aws:s3:::mybucket" + }, + "get_object": { + "key": "myobject", + "size": 252294, + "eTag": "4e9270240d7d62d5ee8dbfcb7a7a3279", + "versionId": "tiogA9Ga7Xi49yfJ6lkeTxPYx7ZK75yn", + "sequencer": "0066A8D0E77DE42BC5" + } + }, + "intelligentTieringEventData": { + "destinationAccessTier": "ARCHIVE_ACCESS" + } + } + ] +} diff --git a/tests/unit/data_classes/required_dependencies/test_s3_intelligent_tiering_event.py b/tests/unit/data_classes/required_dependencies/test_s3_intelligent_tiering_event.py new file mode 100644 index 00000000000..6bae6b8e5d1 --- /dev/null +++ b/tests/unit/data_classes/required_dependencies/test_s3_intelligent_tiering_event.py @@ -0,0 +1,64 @@ +from aws_lambda_powertools.utilities.data_classes import S3Event +from tests.functional.utils import load_event + + +def test_s3_intelligent_tiering_event(): + """Test S3 IntelligentTiering event with get_object field""" + event = S3Event(load_event("s3EventIntelligentTiering.json")) + + # Test first record + record = event.record + assert record.event_name == "IntelligentTiering" + assert record.event_version == "2.3" + assert record.event_source == "aws:s3" + assert record.aws_region == "ap-southeast-2" + + # Test user identity + assert record.user_identity.principal_id == "s3.amazonaws.com" + + # Test S3 object via get_object property (handles both 'object' and 'get_object' keys) + s3_object = record.s3.get_object + assert s3_object.key == "myobject" + assert s3_object.size == 252294 + assert s3_object.etag == "4e9270240d7d62d5ee8dbfcb7a7a3279" + assert s3_object.version_id == "tiogA9Ga7Xi49yfJ6lkeTxPYx7ZK75yn" + assert s3_object.sequencer == "0066A8D0E77DE42BC5" + + # Test bucket + assert record.s3.bucket.name == "mybucket" + assert record.s3.bucket.arn == "arn:aws:s3:::mybucket" + + # Test intelligentTieringEventData + assert record.intelligent_tiering_event_data is not None + assert record.intelligent_tiering_event_data.destination_access_tier == "ARCHIVE_ACCESS" + + # Verify glacierEventData is None + assert record.glacier_event_data is None + + # Test convenience properties + assert event.bucket_name == "mybucket" + assert event.object_key == "myobject" + + +def test_s3_intelligent_tiering_event_iteration(): + """Test iterating through multiple IntelligentTiering records""" + event = S3Event(load_event("s3EventIntelligentTiering.json")) + + records = list(event.records) + assert len(records) == 1 + + for record in event.records: + assert record.event_name == "IntelligentTiering" + assert record.s3.get_object.key == "myobject" + assert record.intelligent_tiering_event_data.destination_access_tier == "ARCHIVE_ACCESS" + + +def test_s3_intelligent_tiering_deep_archive_access(): + """Test IntelligentTiering event with DEEP_ARCHIVE_ACCESS tier""" + raw_event = load_event("s3EventIntelligentTiering.json") + raw_event["Records"][0]["intelligentTieringEventData"]["destinationAccessTier"] = "DEEP_ARCHIVE_ACCESS" + + event = S3Event(raw_event) + record = event.record + + assert record.intelligent_tiering_event_data.destination_access_tier == "DEEP_ARCHIVE_ACCESS" diff --git a/tests/unit/parser/_pydantic/test_s3_intelligent_tiering.py b/tests/unit/parser/_pydantic/test_s3_intelligent_tiering.py new file mode 100644 index 00000000000..801ac87ba8a --- /dev/null +++ b/tests/unit/parser/_pydantic/test_s3_intelligent_tiering.py @@ -0,0 +1,76 @@ +from aws_lambda_powertools.utilities.parser.models import S3Model, S3RecordModel +from tests.functional.utils import load_event + + +def test_s3_intelligent_tiering_event(): + """Test parsing of S3 IntelligentTiering events with get_object field""" + raw_event = load_event("s3EventIntelligentTiering.json") + parsed_event: S3Model = S3Model(**raw_event) + + records = list(parsed_event.Records) + assert len(records) == 1 + + record: S3RecordModel = records[0] + raw_record = raw_event["Records"][0] + + # Verify basic event properties + assert record.eventVersion == "2.3" + assert record.eventSource == "aws:s3" + assert record.awsRegion == "ap-southeast-2" + assert record.eventName == "IntelligentTiering" + + # Verify user identity + user_identity = record.userIdentity + assert user_identity.principalId == "s3.amazonaws.com" + + # Verify request parameters + request_parameters = record.requestParameters + # Note: sourceIPAddress is "s3.amazonaws.com" for IntelligentTiering events, not an IP + assert str(request_parameters.sourceIPAddress) == "s3.amazonaws.com" + + # Verify response elements + assert record.responseElements.x_amz_request_id == raw_record["responseElements"]["x-amz-request-id"] + assert record.responseElements.x_amz_id_2 == raw_record["responseElements"]["x-amz-id-2"] + + # Verify S3 message + s3 = record.s3 + assert s3.s3SchemaVersion == raw_record["s3"]["s3SchemaVersion"] + assert s3.configurationId == raw_record["s3"]["configurationId"] + + # Verify bucket + bucket = s3.bucket + raw_bucket = raw_record["s3"]["bucket"] + assert bucket.name == "mybucket" + assert bucket.ownerIdentity.principalId == raw_bucket["ownerIdentity"]["principalId"] + assert bucket.arn == "arn:aws:s3:::mybucket" + + # Verify get_object field (IntelligentTiering uses 'get_object' instead of 'object') + assert s3.get_object is not None + assert s3.get_object.key == "myobject" + assert s3.get_object.size == 252294 + assert s3.get_object.eTag == "4e9270240d7d62d5ee8dbfcb7a7a3279" + assert s3.get_object.versionId == "tiogA9Ga7Xi49yfJ6lkeTxPYx7ZK75yn" + assert s3.get_object.sequencer == "0066A8D0E77DE42BC5" + + # Verify intelligentTieringEventData + assert record.intelligentTieringEventData is not None + assert record.intelligentTieringEventData.destinationAccessTier == "ARCHIVE_ACCESS" + + # Verify glacierEventData is None for IntelligentTiering events + assert record.glacierEventData is None + + +def test_s3_intelligent_tiering_event_access_tiers(): + """Test different access tier values for IntelligentTiering events""" + raw_event = load_event("s3EventIntelligentTiering.json") + + # Test ARCHIVE_ACCESS tier (from the test event) + parsed_event: S3Model = S3Model(**raw_event) + record = list(parsed_event.Records)[0] + assert record.intelligentTieringEventData.destinationAccessTier == "ARCHIVE_ACCESS" + + # Test DEEP_ARCHIVE_ACCESS tier + raw_event["Records"][0]["intelligentTieringEventData"]["destinationAccessTier"] = "DEEP_ARCHIVE_ACCESS" + parsed_event: S3Model = S3Model(**raw_event) + record = list(parsed_event.Records)[0] + assert record.intelligentTieringEventData.destinationAccessTier == "DEEP_ARCHIVE_ACCESS"