Repository: aws-powertools/powertools-lambda-python

Issue:
Feature request: Enable TumblingWindow support for KinesisStreamEvent
### Use case

Currently KinesisStreamEvent does not support TumblingWindow. That means properties such as state and isFinalInvokeForWindow are not present. I created a data class to support lambda TumblingWindow.

### Solution/User Experience

Propose adding the properties required by Lambda Tumbling Windows.

### Alternative solutions

```markdown

```

### Acknowledgment

- [x] This feature request meets [Powertools for AWS Lambda (Python) Tenets](https://docs.powertools.aws.dev/lambda/python/latest/#tenets)
- [ ] Should this be considered in other Powertools for AWS Lambda languages? i.e. [Java](https://github.com/aws-powertools/powertools-lambda-java/), [TypeScript](https://github.com/aws-powertools/powertools-lambda-typescript/), and [.NET](https://github.com/aws-powertools/powertools-lambda-dotnet/)

Hints from issue comments:
@boring-cyborg: Thanks for opening your first issue here! We'll come back to you as soon as we can.
In the meantime, check out the #python channel on our Powertools for AWS Lambda Discord: [Invite link](https://discord.gg/B8zZKbbyET)
@leandrodamascena: Hey @kiitosu! Thanks for opening this issue! Do you want to open a PR to add this support? I think it's missing in the DynamoDBStreamEvent too.@kiitosu: @leandrodamascena 
Thank you for creating new issue that clarify the problem and solution! I would like to tackle this, and have already commented so at below!
- https://github.com/aws-powertools/powertools-lambda-typescript/issues/3915#issuecomment-2872744755@leandrodamascena: Hi @kiitosu please let me know if you need help submitting a PR! I'll be happy to help and review!

I saw your comment in TypeScript repo and @dreamorosi will assign the issue  for you! 🚀@github-actions: ### ⚠️COMMENT VISIBILITY WARNING⚠️
This issue is now closed. Please be mindful that future comments are hard for our team to see.

If you need more assistance, please either tag a [team member](https://github.com/aws-powertools/powertools-lambda-python/blob/develop/MAINTAINERS.md#current-maintainers) or open a new issue that references this one.

If you wish to keep having a conversation with other community members under this issue feel free to do so.

Solution patch:
diff --git a/aws_lambda_powertools/utilities/data_classes/dynamo_db_stream_event.py b/aws_lambda_powertools/utilities/data_classes/dynamo_db_stream_event.py
index 8da2c983f..d8efbccbb 100644
--- a/aws_lambda_powertools/utilities/data_classes/dynamo_db_stream_event.py
+++ b/aws_lambda_powertools/utilities/data_classes/dynamo_db_stream_event.py
@@ -140,12 +140,25 @@ class DynamoDBRecord(DictWrapper):
         return self.get("userIdentity") or {}
 
 
+class DynamoDBStreamWindow(DictWrapper):
+    @property
+    def start(self) -> str:
+        """The time window started"""
+        return self["start"]
+
+    @property
+    def end(self) -> str:
+        """The time window will end"""
+        return self["end"]
+
+
 class DynamoDBStreamEvent(DictWrapper):
     """Dynamo DB Stream Event
 
     Documentation:
     -------------
     - https://docs.aws.amazon.com/lambda/latest/dg/with-ddb.html
+    - https://docs.aws.amazon.com/lambda/latest/dg/services-ddb-windows.html
 
     Example
     -------
@@ -167,3 +180,30 @@ class DynamoDBStreamEvent(DictWrapper):
     def records(self) -> Iterator[DynamoDBRecord]:
         for record in self["Records"]:
             yield DynamoDBRecord(record)
+
+    @property
+    def window(self) -> DynamoDBStreamWindow | None:
+        window = self.get("window")
+        if window:
+            return DynamoDBStreamWindow(window)
+        return window
+
+    @property
+    def state(self) -> dict[str, Any]:
+        return self.get("state") or {}
+
+    @property
+    def shard_id(self) -> str | None:
+        return self.get("shardId")
+
+    @property
+    def event_source_arn(self) -> str | None:
+        return self.get("eventSourceARN")
+
+    @property
+    def is_final_invoke_for_window(self) -> bool | None:
+        return self.get("isFinalInvokeForWindow")
+
+    @property
+    def is_window_terminated_early(self) -> bool | None:
+        return self.get("isWindowTerminatedEarly")
diff --git a/aws_lambda_powertools/utilities/data_classes/kinesis_stream_event.py b/aws_lambda_powertools/utilities/data_classes/kinesis_stream_event.py
index 6b189f937..64fd26f3a 100644
--- a/aws_lambda_powertools/utilities/data_classes/kinesis_stream_event.py
+++ b/aws_lambda_powertools/utilities/data_classes/kinesis_stream_event.py
@@ -3,7 +3,7 @@ from __future__ import annotations
 import base64
 import json
 import zlib
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from aws_lambda_powertools.utilities.data_classes.cloud_watch_logs_event import (
     CloudWatchLogsDecodedData,
@@ -100,12 +100,25 @@ class KinesisStreamRecord(DictWrapper):
         return KinesisStreamRecordPayload(self["kinesis"])
 
 
+class KinesisStreamWindow(DictWrapper):
+    @property
+    def start(self) -> str:
+        """The time window started"""
+        return self["start"]
+
+    @property
+    def end(self) -> str:
+        """The time window will end"""
+        return self["end"]
+
+
 class KinesisStreamEvent(DictWrapper):
     """Kinesis stream event
 
     Documentation:
     --------------
     - https://docs.aws.amazon.com/lambda/latest/dg/with-kinesis.html
+    - https://docs.aws.amazon.com/lambda/latest/dg/services-kinesis-windows.html
     """
 
     @property
@@ -113,6 +126,33 @@ class KinesisStreamEvent(DictWrapper):
         for record in self["Records"]:
             yield KinesisStreamRecord(record)
 
+    @property
+    def window(self) -> KinesisStreamWindow | None:
+        window = self.get("window")
+        if window:
+            return KinesisStreamWindow(window)
+        return window
+
+    @property
+    def state(self) -> dict[str, Any]:
+        return self.get("state") or {}
+
+    @property
+    def shard_id(self) -> str | None:
+        return self.get("shardId")
+
+    @property
+    def event_source_arn(self) -> str | None:
+        return self.get("eventSourceARN")
+
+    @property
+    def is_final_invoke_for_window(self) -> bool | None:
+        return self.get("isFinalInvokeForWindow")
+
+    @property
+    def is_window_terminated_early(self) -> bool | None:
+        return self.get("isWindowTerminatedEarly")
+
 
 def extract_cloudwatch_logs_from_event(event: KinesisStreamEvent) -> list[CloudWatchLogsDecodedData]:
     return [CloudWatchLogsDecodedData(record.kinesis.data_zlib_compressed_as_json()) for record in event.records]


Test patch:
diff --git a/tests/events/dynamoStreamTumblingWindowEvent.json b/tests/events/dynamoStreamTumblingWindowEvent.json
new file mode 100644
index 000000000..035d08978
--- /dev/null
+++ b/tests/events/dynamoStreamTumblingWindowEvent.json
@@ -0,0 +1,101 @@
+{
+  "Records": [
+    {
+      "eventID": "1",
+      "eventName": "INSERT",
+      "eventVersion": "1.0",
+      "eventSource": "aws:dynamodb",
+      "awsRegion": "us-east-1",
+      "dynamodb": {
+        "Keys": {
+          "Id": {
+            "N": "101"
+          }
+        },
+        "NewImage": {
+          "Message": {
+            "S": "New item!"
+          },
+          "Id": {
+            "N": "101"
+          }
+        },
+        "SequenceNumber": "111",
+        "SizeBytes": 26,
+        "StreamViewType": "NEW_AND_OLD_IMAGES"
+      },
+      "eventSourceARN": "stream-ARN"
+    },
+    {
+      "eventID": "2",
+      "eventName": "MODIFY",
+      "eventVersion": "1.0",
+      "eventSource": "aws:dynamodb",
+      "awsRegion": "us-east-1",
+      "dynamodb": {
+        "Keys": {
+          "Id": {
+            "N": "101"
+          }
+        },
+        "NewImage": {
+          "Message": {
+            "S": "This item has changed"
+          },
+          "Id": {
+            "N": "101"
+          }
+        },
+        "OldImage": {
+          "Message": {
+            "S": "New item!"
+          },
+          "Id": {
+            "N": "101"
+          }
+        },
+        "SequenceNumber": "222",
+        "SizeBytes": 59,
+        "StreamViewType": "NEW_AND_OLD_IMAGES"
+      },
+      "eventSourceARN": "stream-ARN"
+    },
+    {
+      "eventID": "3",
+      "eventName": "REMOVE",
+      "eventVersion": "1.0",
+      "eventSource": "aws:dynamodb",
+      "awsRegion": "us-east-1",
+      "dynamodb": {
+        "Keys": {
+          "Id": {
+            "N": "101"
+          }
+        },
+        "OldImage": {
+          "Message": {
+            "S": "This item has changed"
+          },
+          "Id": {
+            "N": "101"
+          }
+        },
+        "SequenceNumber": "333",
+        "SizeBytes": 38,
+        "StreamViewType": "NEW_AND_OLD_IMAGES"
+      },
+      "eventSourceARN": "stream-ARN"
+    }
+  ],
+  "window": {
+    "start": "2020-07-30T17:00:00Z",
+    "end": "2020-07-30T17:05:00Z"
+  },
+  "state": {
+    "1": "state1"
+  },
+  "shardId": "shard123456789",
+  "eventSourceARN": "stream-ARN",
+  "isFinalInvokeForWindow": false,
+  "isWindowTerminatedEarly": false
+}
diff --git a/tests/events/kinesisStreamEvent.json b/tests/events/kinesisStreamEvent.json
index ef8e20963..cf3a3415e 100644
--- a/tests/events/kinesisStreamEvent.json
+++ b/tests/events/kinesisStreamEvent.json
@@ -32,5 +32,17 @@
       "awsRegion": "us-east-2",
       "eventSourceARN": "arn:aws:kinesis:us-east-2:123456789012:stream/lambda-stream"
     }
-  ]
+  ],
+  "window": {
+    "start": "2020-12-09T07:04:00Z",
+    "end": "2020-12-09T07:06:00Z"
+  },
+  "state": {
+    "1": 282,
+    "2": 715
+  },
+  "shardId": "shardId-000000000006",
+  "eventSourceARN": "arn:aws:kinesis:us-east-1:123456789012:stream/lambda-stream",
+  "isFinalInvokeForWindow": false,
+  "isWindowTerminatedEarly": false
 }
diff --git a/tests/events/kinesisStreamTumblingWindowEvent.json b/tests/events/kinesisStreamTumblingWindowEvent.json
new file mode 100644
index 000000000..020952583
--- /dev/null
+++ b/tests/events/kinesisStreamTumblingWindowEvent.json
@@ -0,0 +1,33 @@
+
+{
+    "Records": [
+        {
+            "kinesis": {
+                "kinesisSchemaVersion": "1.0",
+                "partitionKey": "1",
+                "sequenceNumber": "49590338271490256608559692538361571095921575989136588898",
+                "data": "SGVsbG8sIHRoaXMgaXMgYSB0ZXN0Lg==",
+                "approximateArrivalTimestamp": 1607497475.000
+            },
+            "eventSource": "aws:kinesis",
+            "eventVersion": "1.0",
+            "eventID": "shardId-000000000006:49590338271490256608559692538361571095921575989136588898",
+            "eventName": "aws:kinesis:record",
+            "invokeIdentityArn": "arn:aws:iam::123456789012:role/lambda-kinesis-role",
+            "awsRegion": "us-east-1",
+            "eventSourceARN": "arn:aws:kinesis:us-east-1:123456789012:stream/lambda-stream"
+        }
+    ],
+    "window": {
+        "start": "2020-12-09T07:04:00Z",
+        "end": "2020-12-09T07:06:00Z"
+    },
+    "state": {
+        "1": 282,
+        "2": 715
+    },
+    "shardId": "shardId-000000000006",
+    "eventSourceARN": "arn:aws:kinesis:us-east-1:123456789012:stream/lambda-stream",
+    "isFinalInvokeForWindow": false,
+    "isWindowTerminatedEarly": false
+}
diff --git a/tests/unit/data_classes/required_dependencies/test_dynamo_db_stream_event.py b/tests/unit/data_classes/required_dependencies/test_dynamo_db_stream_event.py
index 8c6b62867..02fdab958 100644
--- a/tests/unit/data_classes/required_dependencies/test_dynamo_db_stream_event.py
+++ b/tests/unit/data_classes/required_dependencies/test_dynamo_db_stream_event.py
@@ -46,6 +46,42 @@ def test_dynamodb_stream_trigger_event():
     assert dynamodb.stream_view_type == StreamViewType.NEW_AND_OLD_IMAGES
 
 
+def test_dynamodb_stream_trigger_with_tumbling_window_event():
+    raw_event = load_event("dynamoStreamTumblingWindowEvent.json")
+    parsed_event = DynamoDBStreamEvent(raw_event)
+
+    records = list(parsed_event.records)
+
+    record = records[0]
+    record_raw = raw_event["Records"][0]
+    assert record.aws_region == record_raw["awsRegion"]
+    assert record.event_id == record_raw["eventID"]
+    assert record.event_name is DynamoDBRecordEventName.INSERT
+    assert record.event_source == record_raw["eventSource"]
+    assert record.event_source_arn == record_raw["eventSourceARN"]
+    assert record.event_version == record_raw["eventVersion"]
+    assert record.user_identity == {}
+    dynamodb = record.dynamodb
+    assert dynamodb is not None
+    keys = dynamodb.keys
+    assert keys is not None
+    assert keys["Id"] == DECIMAL_CONTEXT.create_decimal(101)
+    assert dynamodb.new_image.get("Message") == record_raw["dynamodb"]["NewImage"]["Message"]["S"]
+    assert dynamodb.old_image == {}
+    assert dynamodb.sequence_number == record_raw["dynamodb"]["SequenceNumber"]
+    assert dynamodb.size_bytes == record_raw["dynamodb"]["SizeBytes"]
+    assert dynamodb.stream_view_type == StreamViewType.NEW_AND_OLD_IMAGES
+
+    assert parsed_event.window.raw_event == raw_event["window"]
+    assert parsed_event.window.start == raw_event["window"]["start"]
+    assert parsed_event.window.end == raw_event["window"]["end"]
+    assert parsed_event.state == raw_event["state"]
+    assert parsed_event.shard_id == raw_event["shardId"]
+    assert parsed_event.event_source_arn == raw_event["eventSourceARN"]
+    assert parsed_event.is_final_invoke_for_window == raw_event["isFinalInvokeForWindow"]
+    assert parsed_event.is_window_terminated_early == raw_event["isWindowTerminatedEarly"]
+
+
 def test_dynamodb_stream_record_deserialization_large_int():
     data = {
         "Keys": {"key1": {"attr1": "value1"}},
diff --git a/tests/unit/data_classes/required_dependencies/test_kinesis_stream_event.py b/tests/unit/data_classes/required_dependencies/test_kinesis_stream_event.py
index 5410ed819..2eab5fe90 100644
--- a/tests/unit/data_classes/required_dependencies/test_kinesis_stream_event.py
+++ b/tests/unit/data_classes/required_dependencies/test_kinesis_stream_event.py
@@ -41,6 +41,13 @@ def test_kinesis_stream_event():
     assert kinesis.data_as_bytes() == b"Hello, this is a test."
     assert kinesis.data_as_text() == "Hello, this is a test."
 
+    assert parsed_event.window.raw_event == raw_event["window"]
+    assert parsed_event.state == raw_event["state"]
+    assert parsed_event.shard_id == raw_event["shardId"]
+    assert parsed_event.event_source_arn == raw_event["eventSourceARN"]
+    assert parsed_event.is_final_invoke_for_window == raw_event["isFinalInvokeForWindow"]
+    assert parsed_event.is_window_terminated_early == raw_event["isWindowTerminatedEarly"]
+
 
 def test_kinesis_stream_event_json_data():
     json_value = {"test": "value"}
@@ -56,3 +63,43 @@ def test_kinesis_stream_event_cloudwatch_logs_data_extraction():
     individual_logs = [extract_cloudwatch_logs_from_record(record) for record in event.records]
 
     assert len(extracted_logs) == len(individual_logs)
+
+
+def test_kinesis_stream_with_tumbling_window_event():
+    raw_event = load_event("kinesisStreamTumblingWindowEvent.json")
+    parsed_event = KinesisStreamEvent(raw_event)
+
+    records = list(parsed_event.records)
+    assert len(records) == 1
+    record = records[0]
+
+    record_raw = raw_event["Records"][0]
+
+    assert record.aws_region == record_raw["awsRegion"]
+    assert record.event_id == record_raw["eventID"]
+    assert record.event_name == record_raw["eventName"]
+    assert record.event_source == record_raw["eventSource"]
+    assert record.event_source_arn == record_raw["eventSourceARN"]
+    assert record.event_version == record_raw["eventVersion"]
+    assert record.invoke_identity_arn == record_raw["invokeIdentityArn"]
+
+    kinesis = record.kinesis
+    kinesis_raw = raw_event["Records"][0]["kinesis"]
+
+    assert kinesis.approximate_arrival_timestamp == kinesis_raw["approximateArrivalTimestamp"]
+    assert kinesis.data == kinesis_raw["data"]
+    assert kinesis.kinesis_schema_version == kinesis_raw["kinesisSchemaVersion"]
+    assert kinesis.partition_key == kinesis_raw["partitionKey"]
+    assert kinesis.sequence_number == kinesis_raw["sequenceNumber"]
+
+    assert kinesis.data_as_bytes() == b"Hello, this is a test."
+    assert kinesis.data_as_text() == "Hello, this is a test."
+
+    assert parsed_event.window.raw_event == raw_event["window"]
+    assert parsed_event.window.start == raw_event["window"]["start"]
+    assert parsed_event.window.end == raw_event["window"]["end"]
+    assert parsed_event.state == raw_event["state"]
+    assert parsed_event.shard_id == raw_event["shardId"]
+    assert parsed_event.event_source_arn == raw_event["eventSourceARN"]
+    assert parsed_event.is_final_invoke_for_window == raw_event["isFinalInvokeForWindow"]
+    assert parsed_event.is_window_terminated_early == raw_event["isWindowTerminatedEarly"]