feat(trigger): Kinesis stream event

Add support for Kinesis stream events with a helper method to decode
data

Author: Michael Brewer

aws_lambda_powertools/utilities/trigger/__init__.py

@@ -2,6 +2,7 @@
 from .cloud_watch_logs_event import CloudWatchLogsEvent
 from .dynamo_db_stream_event import DynamoDBStreamEvent
 from .event_bridge_event import EventBridgeEvent
+from .kinesis_stream_event import KinesisStreamEvent
 from .s3_event import S3Event
 from .ses_event import SESEvent
 from .sns_event import SNSEvent
@@ -13,6 +14,7 @@
     "CloudWatchLogsEvent",
     "DynamoDBStreamEvent",
     "EventBridgeEvent",
+    "KinesisStreamEvent",
     "S3Event",
     "SESEvent",
     "SNSEvent",

aws_lambda_powertools/utilities/trigger/kinesis_stream_event.py

@@ -0,0 +1,96 @@
+import base64
+import json
+from typing import Iterator
+
+from aws_lambda_powertools.utilities.trigger.common import DictWrapper
+
+
+class KinesisStreamRecordPayload(DictWrapper):
+    @property
+    def approximate_arrival_timestamp(self) -> float:
+        """The approximate time that the record was inserted into the stream"""
+        return float(self["kinesis"]["approximateArrivalTimestamp"])
+
+    @property
+    def data(self) -> str:
+        """The data blob"""
+        return self["kinesis"]["data"]
+
+    @property
+    def kinesis_schema_version(self) -> str:
+        """Schema version for the record"""
+        return self["kinesis"]["kinesisSchemaVersion"]
+
+    @property
+    def partition_key(self) -> str:
+        """Identifies which shard in the stream the data record is assigned to"""
+        return self["kinesis"]["partitionKey"]
+
+    @property
+    def sequence_number(self) -> str:
+        """The unique identifier of the record within its shard"""
+        return self["kinesis"]["sequenceNumber"]
+
+    def data_as_text(self) -> str:
+        """Decode binary encoded data as text"""
+        return base64.b64decode(self.data).decode("utf-8")
+
+    def data_as_json(self) -> dict:
+        """Decode binary encoded data as json"""
+        return json.loads(self.data_as_text())
+
+
+class KinesisStreamRecord(DictWrapper):
+    @property
+    def aws_region(self) -> str:
+        """AWS region where the event originated eg: us-east-1"""
+        return self["awsRegion"]
+
+    @property
+    def event_id(self) -> str:
+        """A globally unique identifier for the event that was recorded in this stream record."""
+        return self["eventID"]
+
+    @property
+    def event_name(self) -> str:
+        """Event type eg: aws:kinesis:record"""
+        return self["eventName"]
+
+    @property
+    def event_source(self) -> str:
+        """The AWS service from which the Kinesis event originated. For Kinesis, this is aws:kinesis"""
+        return self["eventSource"]
+
+    @property
+    def event_source_arn(self) -> str:
+        """The Amazon Resource Name (ARN) of the event source"""
+        return self["eventSourceARN"]
+
+    @property
+    def event_version(self) -> str:
+        """The eventVersion key value contains a major and minor version in the form <major>.<minor>."""
+        return self["eventVersion"]
+
+    @property
+    def invoke_identity_arn(self) -> str:
+        """The ARN for the identity used to invoke the Lambda Function"""
+        return self["invokeIdentityArn"]
+
+    @property
+    def kinesis(self) -> KinesisStreamRecordPayload:
+        """Underlying Kinesis record associated with the event"""
+        return KinesisStreamRecordPayload(self._data)
+
+
+class KinesisStreamEvent(dict):
+    """Kinesis stream event
+
+    Documentation:
+    --------------
+    - https://docs.aws.amazon.com/lambda/latest/dg/with-kinesis.html
+    """
+
+    @property
+    def records(self) -> Iterator[KinesisStreamRecord]:
+        for record in self["Records"]:
+            yield KinesisStreamRecord(record)

aws_lambda_powertools/utilities/trigger/ses_event.py

@@ -184,6 +184,7 @@ def event_source(self) -> str:
 
     @property
     def event_version(self) -> str:
+        """The eventVersion key value contains a major and minor version in the form <major>.<minor>."""
         return self["eventVersion"]
 
     @property

tests/events/kinesisStreamEvent.json

@@ -0,0 +1,36 @@
+{
+  "Records": [
+    {
+      "kinesis": {
+        "kinesisSchemaVersion": "1.0",
+        "partitionKey": "1",
+        "sequenceNumber": "49590338271490256608559692538361571095921575989136588898",
+        "data": "SGVsbG8sIHRoaXMgaXMgYSB0ZXN0Lg==",
+        "approximateArrivalTimestamp": 1545084650.987
+      },
+      "eventSource": "aws:kinesis",
+      "eventVersion": "1.0",
+      "eventID": "shardId-000000000006:49590338271490256608559692538361571095921575989136588898",
+      "eventName": "aws:kinesis:record",
+      "invokeIdentityArn": "arn:aws:iam::123456789012:role/lambda-role",
+      "awsRegion": "us-east-2",
+      "eventSourceARN": "arn:aws:kinesis:us-east-2:123456789012:stream/lambda-stream"
+    },
+    {
+      "kinesis": {
+        "kinesisSchemaVersion": "1.0",
+        "partitionKey": "1",
+        "sequenceNumber": "49590338271490256608559692540925702759324208523137515618",
+        "data": "VGhpcyBpcyBvbmx5IGEgdGVzdC4=",
+        "approximateArrivalTimestamp": 1545084711.166
+      },
+      "eventSource": "aws:kinesis",
+      "eventVersion": "1.0",
+      "eventID": "shardId-000000000006:49590338271490256608559692540925702759324208523137515618",
+      "eventName": "aws:kinesis:record",
+      "invokeIdentityArn": "arn:aws:iam::123456789012:role/lambda-role",
+      "awsRegion": "us-east-2",
+      "eventSourceARN": "arn:aws:kinesis:us-east-2:123456789012:stream/lambda-stream"
+    }
+  ]
+}

tests/functional/test_lambda_trigger_events.py

@@ -1,3 +1,4 @@
+import base64
 import json
 import os
 from secrets import compare_digest
@@ -7,6 +8,7 @@
     APIGatewayProxyEventV2,
     CloudWatchLogsEvent,
     EventBridgeEvent,
+    KinesisStreamEvent,
     S3Event,
     SESEvent,
     SNSEvent,
@@ -603,3 +605,37 @@ def test_api_gateway_proxy_v2_get_header_value():
 
     value = event.get_header_value("unknown")
     assert value is None
+
+
+def test_kinesis_stream_event():
+    event = KinesisStreamEvent(load_event("kinesisStreamEvent.json"))
+
+    records = list(event.records)
+    assert len(records) == 2
+    record = records[0]
+
+    assert record.aws_region == "us-east-2"
+    assert record.event_id == "shardId-000000000006:49590338271490256608559692538361571095921575989136588898"
+    assert record.event_name == "aws:kinesis:record"
+    assert record.event_source == "aws:kinesis"
+    assert record.event_source_arn == "arn:aws:kinesis:us-east-2:123456789012:stream/lambda-stream"
+    assert record.event_version == "1.0"
+    assert record.invoke_identity_arn == "arn:aws:iam::123456789012:role/lambda-role"
+
+    kinesis = record.kinesis
+    assert kinesis._data["kinesis"] == event["Records"][0]["kinesis"]
+
+    assert kinesis.approximate_arrival_timestamp == 1545084650.987
+    assert kinesis.data == event["Records"][0]["kinesis"]["data"]
+    assert kinesis.kinesis_schema_version == "1.0"
+    assert kinesis.partition_key == "1"
+    assert kinesis.sequence_number == "49590338271490256608559692538361571095921575989136588898"
+
+    assert kinesis.data_as_text() == "Hello, this is a test."
+
+
+def test_kinesis_stream_event_json_data():
+    json_value = {"test": "value"}
+    data = base64.b64encode(bytes(json.dumps(json_value), "utf-8")).decode("utf-8")
+    event = KinesisStreamEvent({"Records": [{"kinesis": {"data": data}}]})
+    assert next(event.records).kinesis.data_as_json() == json_value