Adding parser

Author: leandrodamascena

aws_lambda_powertools/utilities/parser/models/__init__.py

@@ -68,6 +68,7 @@
     S3Model,
     S3RecordModel,
 )
+from .s3_batch_operation import S3BatchOperationJobModel, S3BatchOperationModel, S3BatchOperationTaskModel
 from .s3_event_notification import (
     S3SqsEventNotificationModel,
     S3SqsEventNotificationRecordModel,
@@ -177,4 +178,7 @@
     "BedrockAgentEventModel",
     "BedrockAgentRequestBodyModel",
     "BedrockAgentRequestMediaModel",
+    "S3BatchOperationJobModel",
+    "S3BatchOperationModel",
+    "S3BatchOperationTaskModel",
 ]

aws_lambda_powertools/utilities/parser/models/s3_batch_operation.py

@@ -0,0 +1,34 @@
+from typing import Any, Dict, List, Optional
+
+from pydantic import BaseModel, validator
+
+from aws_lambda_powertools.utilities.parser.types import Literal
+
+
+class S3BatchOperationTaskModel(BaseModel):
+    taskId: str
+    s3Key: str
+    s3VersionId: Optional[str] = None
+    s3BucketArn: Optional[str] = None
+    s3Bucket: Optional[str] = None
+
+    @validator("s3Bucket", pre=True, always=True)
+    def validate_bucket(cls, current_value, values):
+        # Get the s3 bucket, either from 's3Bucket' property (invocationSchemaVersion '2.0')
+        # or from 's3BucketArn' (invocationSchemaVersion '1.0')
+        if values.get("s3BucketArn") and not current_value:
+            # Replace s3Bucket value with the value from s3BucketArn
+            return values["s3BucketArn"].split(":::")[-1]
+        return current_value
+
+
+class S3BatchOperationJobModel(BaseModel):
+    id: str
+    userArguments: Optional[Dict[str, Any]] = None
+
+
+class S3BatchOperationModel(BaseModel):
+    invocationId: str
+    invocationSchemaVersion: Literal["1.0", "2.0"]
+    job: S3BatchOperationJobModel
+    tasks: List[S3BatchOperationTaskModel]

docs/utilities/parser.md

@@ -191,6 +191,7 @@ Parser comes with the following built-in models:
 | **KinesisFirehoseModel**                    | Lambda Event Source payload for Amazon Kinesis Firehose                               |
 | **KinesisFirehoseSqsModel**                 | Lambda Event Source payload for SQS messages wrapped in Kinesis Firehose records      |
 | **LambdaFunctionUrlModel**                  | Lambda Event Source payload for Lambda Function URL payload                           |
+| **S3BatchOperationModel**                   | Lambda Event Source payload for Amazon S3 Batch Operation                             |
 | **S3EventNotificationEventBridgeModel**     | Lambda Event Source payload for Amazon S3 Event Notification to EventBridge.          |
 | **S3Model**                                 | Lambda Event Source payload for Amazon S3                                             |
 | **S3ObjectLambdaEvent**                     | Lambda Event Source payload for Amazon S3 Object Lambda                               |

tests/unit/parser/test_s3_batch_operation.py

@@ -0,0 +1,38 @@
+from aws_lambda_powertools.utilities.parser.models import S3BatchOperationModel
+from tests.functional.utils import load_event
+
+
+def test_s3_batch_operation_v1_trigger_event():
+    raw_event = load_event("s3BatchOperationEventSchemaV1.json")
+    parsed_event: S3BatchOperationModel = S3BatchOperationModel(**raw_event)
+
+    tasks = list(parsed_event.tasks)
+    assert len(tasks) == 1
+
+    assert parsed_event.invocationId == raw_event["invocationId"]
+    assert parsed_event.invocationSchemaVersion == raw_event["invocationSchemaVersion"]
+    assert parsed_event.job.id == raw_event["job"]["id"]
+
+    assert tasks[0].taskId == raw_event["tasks"][0]["taskId"]
+    assert tasks[0].s3Key == raw_event["tasks"][0]["s3Key"]
+    assert tasks[0].s3VersionId == raw_event["tasks"][0]["s3VersionId"]
+    assert tasks[0].s3BucketArn == raw_event["tasks"][0]["s3BucketArn"]
+    assert tasks[0].s3Bucket == "powertools-dataset"
+
+
+def test_s3_batch_operation_v2_trigger_event():
+    raw_event = load_event("s3BatchOperationEventSchemaV2.json")
+    parsed_event: S3BatchOperationModel = S3BatchOperationModel(**raw_event)
+
+    tasks = list(parsed_event.tasks)
+    assert len(tasks) == 1
+
+    assert parsed_event.invocationId == raw_event["invocationId"]
+    assert parsed_event.invocationSchemaVersion == raw_event["invocationSchemaVersion"]
+    assert parsed_event.job.id == raw_event["job"]["id"]
+    assert parsed_event.job.userArguments == raw_event["job"]["userArguments"]
+
+    assert tasks[0].taskId == raw_event["tasks"][0]["taskId"]
+    assert tasks[0].s3Key == raw_event["tasks"][0]["s3Key"]
+    assert tasks[0].s3VersionId == raw_event["tasks"][0]["s3VersionId"]
+    assert tasks[0].s3Bucket == raw_event["tasks"][0]["s3Bucket"]