feat: Add Kinesis lambda event support to Parser utility

Author: ran-isenberg

aws_lambda_powertools/utilities/parser/envelopes/__init__.py

@@ -1,14 +1,14 @@
 from .base import BaseEnvelope
 from .dynamodb import DynamoDBStreamEnvelope
 from .event_bridge import EventBridgeEnvelope
-from .kinesis import KinesisEnvelope
+from .kinesis import KinesisDataStreamEnvelope
 from .sns import SnsEnvelope
 from .sqs import SqsEnvelope
 
 __all__ = [
     "DynamoDBStreamEnvelope",
     "EventBridgeEnvelope",
-    "KinesisEnvelope",
+    "KinesisDataStreamEnvelope",
     "SnsEnvelope",
     "SqsEnvelope",
     "BaseEnvelope",

aws_lambda_powertools/utilities/parser/envelopes/kinesis.py

@@ -8,8 +8,8 @@
 logger = logging.getLogger(__name__)
 
 
-class KinesisEnvelope(BaseEnvelope):
-    """Kinesis Envelope to extract array of Records
+class KinesisDataStreamEnvelope(BaseEnvelope):
+    """Kinesis Data Stream Envelope to extract array of Records
 
     The record's data parameter is a base64 encoded string which is parsed into a bytes array, 
     though it can also be a JSON encoded string.

aws_lambda_powertools/utilities/parser/models/__init__.py

@@ -1,6 +1,6 @@
 from .dynamodb import DynamoDBStreamChangedRecordModel, DynamoDBStreamModel, DynamoDBStreamRecordModel
 from .event_bridge import EventBridgeModel
-from .kinesis import KinesisStreamModel, KinesisStreamRecord, KinesisStreamRecordPayload
+from .kinesis import KinesisDataStreamRecordPayload, KinesisStreamModel, KinesisStreamRecord
 from .ses import SesModel, SesRecordModel
 from .sns import SnsModel, SnsNotificationModel, SnsRecordModel
 from .sqs import SqsModel, SqsRecordModel
@@ -12,7 +12,7 @@
     "DynamoDBStreamRecordModel",
     "KinesisStreamModel",
     "KinesisStreamRecord",
-    "KinesisStreamRecordPayload",
+    "KinesisDataStreamRecordPayload",
     "SesModel",
     "SesRecordModel",
     "SnsModel",

aws_lambda_powertools/utilities/parser/models/kinesis.py

@@ -1,13 +1,16 @@
 import base64
+import logging
 from binascii import Error as BinAsciiError
 from typing import List
 
 from pydantic import BaseModel, validator
 from pydantic.types import PositiveInt
 from typing_extensions import Literal
 
+logger = logging.getLogger(__name__)
 
-class KinesisStreamRecordPayload(BaseModel):
+
+class KinesisDataStreamRecordPayload(BaseModel):
     kinesisSchemaVersion: str
     partitionKey: str
     sequenceNumber: PositiveInt
@@ -17,6 +20,7 @@ class KinesisStreamRecordPayload(BaseModel):
     @validator("data", pre=True)
     def data_base64_decode(cls, value):
         try:
+            logger.debug("Decoding base64 Kinesis data record before parsing")
             return base64.b64decode(value)
         except (BinAsciiError, TypeError):
             raise ValueError("base64 decode failed")
@@ -30,7 +34,7 @@ class KinesisStreamRecord(BaseModel):
     invokeIdentityArn: str
     awsRegion: str
     eventSourceARN: str
-    kinesis: KinesisStreamRecordPayload
+    kinesis: KinesisDataStreamRecordPayload
 
 
 class KinesisStreamModel(BaseModel):

tests/functional/parser/test_kinesis.py

@@ -3,13 +3,13 @@
 import pytest
 
 from aws_lambda_powertools.utilities.parser import ValidationError, envelopes, event_parser
-from aws_lambda_powertools.utilities.parser.models import KinesisStreamModel, KinesisStreamRecordPayload
+from aws_lambda_powertools.utilities.parser.models import KinesisDataStreamRecordPayload, KinesisStreamModel
 from aws_lambda_powertools.utilities.typing import LambdaContext
 from tests.functional.parser.schemas import MyKinesisBusiness
 from tests.functional.parser.utils import load_event
 
 
-@event_parser(model=MyKinesisBusiness, envelope=envelopes.KinesisEnvelope)
+@event_parser(model=MyKinesisBusiness, envelope=envelopes.KinesisDataStreamEnvelope)
 def handle_kinesis(event: List[MyKinesisBusiness], _: LambdaContext):
     assert len(event) == 1
     record: KinesisStreamModel = event[0]
@@ -31,7 +31,7 @@ def handle_kinesis_no_envelope(event: KinesisStreamModel, _: LambdaContext):
     assert record.eventVersion == "1.0"
     assert record.invokeIdentityArn == "arn:aws:iam::123456789012:role/lambda-role"
 
-    kinesis: KinesisStreamRecordPayload = record.kinesis
+    kinesis: KinesisDataStreamRecordPayload = record.kinesis
     assert kinesis.approximateArrivalTimestamp == 1545084650.987
     assert kinesis.kinesisSchemaVersion == "1.0"
     assert kinesis.partitionKey == "1"