feat(parser): add model for the DynamoDB Stream Lambda invocation record

This commit adds a model for the invocation record that a Lambda with
a DynamoDB event source sends to a SNS, SQS or S3 destination for
failed invocations. The record format is described in

https://docs.aws.amazon.com/lambda/latest/dg/services-dynamodb-errors.html

The model can be used in the handler of a lambda that processes failed
invocations from SNS or SQS by combining it with the source envelope
model, e.g.,

  from aws_lambda_powertools.utilities.parser import envelopes, event_parser
  from aws_lambda_powertools.utilities.parser.models import DynamoDBStreamLambdaInvocationModel

  @event_parser(model=DDBStreamInvocationModel, envelope=envelopes.SqsEnvelope)
  def lambda_handler(event: list[DDBStreamInvocationModel], context: LambdaContext):
      ...

Author: exg

aws_lambda_powertools/utilities/parser/models/__init__.py

@@ -66,6 +66,7 @@
 )
 from .dynamodb import (
     DynamoDBStreamChangedRecordModel,
+    DynamoDBStreamLambdaInvocationModel,
     DynamoDBStreamModel,
     DynamoDBStreamRecordModel,
 )
@@ -173,6 +174,7 @@
     "DynamoDBStreamModel",
     "EventBridgeModel",
     "DynamoDBStreamChangedRecordModel",
+    "DynamoDBStreamLambdaInvocationModel",
     "DynamoDBStreamRecordModel",
     "DynamoDBStreamChangedRecordModel",
     "KinesisDataStreamModel",

aws_lambda_powertools/utilities/parser/models/dynamodb.py

@@ -110,3 +110,83 @@ class DynamoDBStreamModel(BaseModel):
             },
         ],
     )
+
+
+class DDBStreamBatchInfo(BaseModel):
+    approximateArrivalOfFirstRecord: datetime = Field(
+        description="The approximate date and time when the first stream record from the batch was created"
+        ", in ISO-8601 format.",
+        examples=["1970-01-01T00:00:00.000Z"],
+    )
+    approximateArrivalOfLastRecord: datetime = Field(
+        description="The approximate date and time when the last stream record from the batch was created"
+        ", in ISO-8601 format.",
+        examples=["1970-01-01T00:00:00.000Z"],
+    )
+    batchSize: int = Field(
+        description="The size of the batch.",
+        examples=[1],
+    )
+    endSequenceNumber: str = Field(
+        description="The unique identifier of the last stream record from the batch.",
+        examples=["222"],
+    )
+    shardId: str = Field(
+        description="The unique identifier of the DynamoDB Stream shard that contains the records from the batch.",
+        examples=["shardId-00000000000000000000-00000000"],
+    )
+    startSequenceNumber: str = Field(
+        description="The unique identifier of the first stream record from the batch.",
+        examples=["222"],
+    )
+    streamArn: str = Field(
+        description="The Amazon Resource Name (ARN) of the DynamoDB stream.",
+        examples=["arn:aws:dynamodb:us-west-2:123456789012:table/ExampleTable/stream/2021-01-01T00:00:00.000"],
+    )
+
+
+class RequestContext(BaseModel):
+    approximateInvokeCount: int = Field(
+        description="The number of Lambda invocations for the record.",
+        examples=[1],
+    )
+    condition: str = Field(
+        description="The condition that caused the record to be discarded.",
+        examples=["RetryAttemptsExhausted"],
+    )
+    functionArn: str = Field(
+        description="The Amazon Resource Name (ARN) of the Lambda.",
+        examples=["arn:aws:lambda:eu-west-1:809313241:function:test"],
+    )
+    requestId: str = Field(
+        description="The unique identifier of the request.",
+    )
+
+
+class ResponseContext(BaseModel):
+    executedVersion: str = Field(
+        description="The version of the Lambda executed",
+        examples=["$LATEST"],
+    )
+    functionError: str = Field(
+        description="",
+        examples=["Unhandled"],
+    )
+    statusCode: int = Field(
+        description="The status code returned by the Lambda",
+    )
+
+
+# https://docs.aws.amazon.com/lambda/latest/dg/services-dynamodb-errors.html
+class DynamoDBStreamLambdaInvocationModel(BaseModel):
+    DDBStreamBatchInfo: DDBStreamBatchInfo
+    requestContext: RequestContext
+    responseContext: ResponseContext
+    timestamp: datetime = Field(
+        description="The record time, in ISO-8601 format.",
+        examples=["1970-01-01T00:00:00.000Z"],
+    )
+    version: str = Field(
+        description="The version of the record format.",
+        examples=["1.0"],
+    )

docs/utilities/parser.md

@@ -133,6 +133,7 @@ The example above uses `SqsModel`. Other built-in models can be found below.
 | **CognitoCreateAuthChallengeTriggerModel**  | Lambda User Pool Create Auth Challenge trigger event                                          |
 | **CognitoVerifyAuthChallengeTriggerModel**  | Lambda User Pool Verify Auth Challenge trigger event                                          |
 | **DynamoDBStreamModel**                     | Lambda Event Source payload for Amazon DynamoDB Streams                                       |
+| **DynamoDBStreamLambdaInvocationModel**     | Lambda invocation record for Amazon DynamoDB Streams
 | **EventBridgeModel**                        | Lambda Event Source payload for Amazon EventBridge                                            |
 | **IoTCoreThingEvent**                       | Lambda Event Source payload for IoT Core Thing created, updated, or deleted.                  |
 | **IoTCoreThingTypeEvent**                   | Lambda Event Source payload for IoT Core Thing Type events.                                   |

tests/events/dynamoStreamLambdaInvocationEvent.json

@@ -0,0 +1,24 @@
+{
+  "requestContext": {
+    "requestId": "316aa6d0-8154-xmpl-9af7-85d5f4a6bc81",
+    "functionArn": "arn:aws:lambda:us-east-2:123456789012:function:myfunction",
+    "condition": "RetryAttemptsExhausted",
+    "approximateInvokeCount": 1
+  },
+  "responseContext": {
+    "statusCode": 200,
+    "executedVersion": "$LATEST",
+    "functionError": "Unhandled"
+  },
+  "version": "1.0",
+  "timestamp": "2019-11-14T00:13:49Z",
+  "DDBStreamBatchInfo": {
+    "shardId": "shardId-00000001573689847184-864758bb",
+    "startSequenceNumber": "800000000003126276362",
+    "endSequenceNumber": "800000000003126276362",
+    "approximateArrivalOfFirstRecord": "2019-11-14T00:13:19Z",
+    "approximateArrivalOfLastRecord": "2019-11-14T00:13:19Z",
+    "batchSize": 1,
+    "streamArn": "arn:aws:dynamodb:us-east-2:123456789012:table/mytable/stream/2019-11-14T00:04:06.388"
+  }
+}

tests/unit/parser/_pydantic/schemas.py

@@ -4,6 +4,7 @@
 
 from aws_lambda_powertools.utilities.parser.models import (
     DynamoDBStreamChangedRecordModel,
+    DynamoDBStreamLambdaInvocationModel,
     DynamoDBStreamModel,
     DynamoDBStreamRecordModel,
     EventBridgeModel,

tests/unit/parser/_pydantic/test_dynamodb.py

@@ -1,6 +1,7 @@
 import pytest
 
 from aws_lambda_powertools.utilities.parser import ValidationError, envelopes, parse
+from aws_lambda_powertools.utilities.parser.models import DynamoDBStreamLambdaInvocationModel
 from tests.functional.utils import load_event
 from tests.unit.parser._pydantic.schemas import MyAdvancedDynamoBusiness, MyDynamoBusiness
 
@@ -83,3 +84,28 @@ def test_validate_event_does_not_conform_with_model():
     raw_event: dict = {"hello": "s"}
     with pytest.raises(ValidationError):
         parse(event=raw_event, model=MyDynamoBusiness, envelope=envelopes.DynamoDBStreamEnvelope)
+
+
+def test_dynamo_db_stream_lambda_invocation_event():
+    raw_event = load_event("dynamoStreamLambdaInvocationEvent.json")
+    parsed_event: DynamoDBStreamLambdaInvocationModel = parse(
+        event=raw_event,
+        model=DynamoDBStreamLambdaInvocationModel,
+    )
+    assert (
+        parsed_event.DDBStreamBatchInfo.approximateArrivalOfFirstRecord.strftime("%Y-%m-%dT%H:%M:%SZ")
+        == raw_event["DDBStreamBatchInfo"]["approximateArrivalOfFirstRecord"]
+    )
+    assert (
+        parsed_event.DDBStreamBatchInfo.approximateArrivalOfLastRecord.strftime("%Y-%m-%dT%H:%M:%SZ")
+        == raw_event["DDBStreamBatchInfo"]["approximateArrivalOfLastRecord"]
+    )
+    assert parsed_event.DDBStreamBatchInfo.batchSize == raw_event["DDBStreamBatchInfo"]["batchSize"]
+    assert parsed_event.DDBStreamBatchInfo.endSequenceNumber == raw_event["DDBStreamBatchInfo"]["endSequenceNumber"]
+    assert parsed_event.DDBStreamBatchInfo.shardId == raw_event["DDBStreamBatchInfo"]["shardId"]
+    assert parsed_event.DDBStreamBatchInfo.startSequenceNumber == raw_event["DDBStreamBatchInfo"]["startSequenceNumber"]
+    assert parsed_event.DDBStreamBatchInfo.streamArn == raw_event["DDBStreamBatchInfo"]["streamArn"]
+    assert parsed_event.requestContext.model_dump() == raw_event["requestContext"]
+    assert parsed_event.responseContext.model_dump() == raw_event["responseContext"]
+    assert parsed_event.timestamp.strftime("%Y-%m-%dT%H:%M:%SZ") == raw_event["timestamp"]
+    assert parsed_event.version == raw_event["version"]