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

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 DynamoDBStreamLambdaOnFailureDestinationModel

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

Co-authored-by: Leandro Damascena <[email protected]>

Author: exg

aws_lambda_powertools/utilities/parser/models/__init__.py

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

aws_lambda_powertools/utilities/parser/models/dynamodb.py

@@ -2,7 +2,8 @@
 from datetime import datetime
 from typing import Any, Dict, List, Literal, Optional, Type, Union
 
-from pydantic import BaseModel, Field, field_validator
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+from pydantic.alias_generators import to_camel
 
 from aws_lambda_powertools.shared.dynamodb_deserializer import TypeDeserializer
 
@@ -110,3 +111,91 @@ class DynamoDBStreamModel(BaseModel):
             },
         ],
     )
+
+
+class DDBStreamBatchInfo(BaseModel):
+    model_config = ConfigDict(alias_generator=to_camel)
+
+    approximate_arrival_of_first_record: 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"],
+    )
+    approximate_arrival_of_last_record: 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"],
+    )
+    batch_size: int = Field(
+        description="The size of the batch.",
+        examples=[1],
+    )
+    end_sequence_number: str = Field(
+        description="The unique identifier of the last stream record from the batch.",
+        examples=["222"],
+    )
+    shard_id: str = Field(
+        description="The unique identifier of the DynamoDB Stream shard that contains the records from the batch.",
+        examples=["shardId-00000000000000000000-00000000"],
+    )
+    start_sequence_number: str = Field(
+        description="The unique identifier of the first stream record from the batch.",
+        examples=["222"],
+    )
+    stream_arn: 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):
+    model_config = ConfigDict(alias_generator=to_camel)
+
+    approximate_invoke_count: 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"],
+    )
+    function_arn: str = Field(
+        description="The Amazon Resource Name (ARN) of the Lambda.",
+        examples=["arn:aws:lambda:eu-west-1:809313241:function:test"],
+    )
+    request_id: str = Field(
+        description="The unique identifier of the request.",
+    )
+
+
+class ResponseContext(BaseModel):
+    model_config = ConfigDict(alias_generator=to_camel)
+
+    executed_version: str = Field(
+        description="The version of the Lambda executed",
+        examples=["$LATEST"],
+    )
+    function_error: str = Field(
+        description="",
+        examples=["Unhandled"],
+    )
+    status_code: int = Field(
+        description="The status code returned by the Lambda",
+    )
+
+
+# https://docs.aws.amazon.com/lambda/latest/dg/services-dynamodb-errors.html
+class DynamoDBStreamLambdaOnFailureDestinationModel(BaseModel):
+    model_config = ConfigDict(alias_generator=to_camel)
+
+    ddb_stream_batch_info: DDBStreamBatchInfo = Field(alias="DDBStreamBatchInfo")
+    request_context: RequestContext
+    response_context: 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"],
+    )