fix typo, make asdict a function

Author: roger-zhangg

aws_lambda_powertools/utilities/data_classes/common.py

@@ -15,7 +15,7 @@ class DictWrapper(Mapping):
 
     def __init__(self, data: Dict[str, Any], json_deserializer: Optional[Callable] = None):
         """
-        ParametersW
+        Parameters
         ----------
         data : Dict[str, Any]
             Lambda Event Source Event payload

aws_lambda_powertools/utilities/data_classes/kinesis_firehose_event.py

@@ -1,15 +1,11 @@
 import base64
 import json
-import sys
 from dataclasses import dataclass
 from typing import Any, Callable, Dict, Iterator, List, Optional
 
-from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+from typing_extensions import Literal
 
-if sys.version_info >= (3, 8):
-    from typing import Literal
-else:
-    from typing_extensions import Literal
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
 
 
 @dataclass
@@ -22,7 +18,6 @@ class KinesisFirehoseResponseRecordMetadata:
 
     partition_keys: Optional[Dict[str, str]]
 
-    @property
     def asdict(self) -> Optional[Dict]:
         if self.partition_keys is not None:
             return {"partitionKeys": self.partition_keys}
@@ -42,7 +37,8 @@ class KinesisFirehoseResponseRecord:
     record_id: str
     """processing result, supported value: Ok, Dropped, ProcessingFailed"""
     result: Literal["Ok", "Dropped", "ProcessingFailed"]
-    """The data blob, base64-encoded, optional at init"""
+    """data blob, base64-encoded, optional at init. Allows pass in base64-encoded data directly or
+            use either function like `data_from_text`, `data_from_json` to populate data"""
     data: Optional[str] = None
     """Optional: metadata associated with this record; present only when Kinesis Stream is source"""
     metadata: Optional[KinesisFirehoseResponseRecordMetadata] = None
@@ -63,15 +59,14 @@ def data_from_json(self, data: Any):
         """Populate data field using any structure that could be converted to json"""
         self.data_from_text(data=self.json_serializer(data))
 
-    @property
     def asdict(self) -> Dict:
         r: Dict[str, Any] = {
             "recordId": self.record_id,
             "result": self.result,
             "data": self.data,
         }
         if self.metadata:
-            r["metadata"] = self.metadata.asdict
+            r["metadata"] = self.metadata.asdict()
         return r
 
     @property
@@ -115,11 +110,12 @@ def add_record(self, record: KinesisFirehoseResponseRecord):
         else:
             self.records = [record]
 
-    @property
     def asdict(self) -> Dict:
+        # make sure return size is less than 6MB
         if not self.records:
             return {}
-        return {"records": [r.asdict for r in self.records]}
+
+        return {"records": [r.asdict() for r in self.records]}
 
 
 class KinesisFirehoseRecordMetadata(DictWrapper):
@@ -200,6 +196,16 @@ def create_firehose_response_record(
         result: Literal["Ok", "Dropped", "ProcessingFailed"],
         data: Optional[str] = None,
     ) -> KinesisFirehoseResponseRecord:
+        """create a KinesisFirehoseResponseRecord directly using the record_id and given values
+        Parameters
+        ----------
+        result : Literal["Ok", "Dropped", "ProcessingFailed"]
+            processing result, supported value: Ok, Dropped, ProcessingFailed
+        data : str, optional
+            data blob, base64-encoded, optional at init. Allows pass in base64-encoded data directly or
+            use either function like `data_from_text`, `data_from_json` to populate data
+
+        """
         return KinesisFirehoseResponseRecord(record_id=self.record_id, result=result, data=data)
 
 

docs/utilities/data_classes.md

@@ -981,12 +981,23 @@ inline, and re-emit them back to the Delivery Stream.
 Similar to Kinesis Data Streams, the events contain base64 encoded data. You can use the helper
 function to access the data either as json or plain text, depending on the original payload.
 
+When constructing response to Firehose, You can utilize the `KinesisFirehoseResponse` class shown
+in the example below.
+
 === "app.py"
 
     ```python
     --8<-- "examples/event_sources/src/kinesis_firehose_delivery_stream.py"
     ```
 
+You can also construct response without using `event_source` wrapper. Shown in the example below.
+
+=== "app.py"
+
+    ```python
+    --8<-- "examples/event_sources/src/kinesis_firehose_response.py"
+    ```
+
 ### Lambda Function URL
 
 === "app.py"

examples/event_sources/src/kinesis_firehose_delivery_stream.py

@@ -15,11 +15,13 @@ def lambda_handler(event: KinesisFirehoseEvent, context: LambdaContext):
         data = record.data_as_json
 
         ## do all kind of stuff with data
+        ## generate data to return
+        new_data = {"tool_used": "powertools_dataclass", "original_payload": data}
 
         processed_record = record.create_firehose_response_record(result="Ok")
-        processed_record.data_from_json(data=data)
+        processed_record.data_from_json(data=new_data)
 
         result.add_record(processed_record)
 
     # return transformed records
-    return result.asdict
+    return result.asdict()

examples/event_sources/src/kinesis_firehose_response.py

@@ -0,0 +1,24 @@
+import base64
+
+from aws_lambda_powertools.utilities.data_classes import (
+    KinesisFirehoseResponse,
+    KinesisFirehoseResponseRecord,
+)
+
+
+def lambda_handler(event, context):
+    result = KinesisFirehoseResponse()
+
+    for record in event["records"]:
+        print(record["recordId"])
+        payload = base64.b64decode(record["data"]).decode("utf-8")
+        ## do all kind of stuff with payload
+        ## generate data to return
+        new_data = {"tool_used": "powertools_dataclass", "original_payload": payload}
+
+        processed_record = KinesisFirehoseResponseRecord(record_id=record["recordId"], result="Ok")
+        processed_record.data_from_text(data=new_data)
+        result.add_record(processed_record)
+
+    # return transformed records
+    return result.asdict()

tests/unit/data_classes/test_kinesis_firehose_response.py

@@ -2,6 +2,7 @@
     KinesisFirehoseEvent,
     KinesisFirehoseResponse,
     KinesisFirehoseResponseRecord,
+    KinesisFirehoseResponseRecordMetadata,
 )
 from tests.functional.utils import load_event
 
@@ -15,13 +16,15 @@ def test_kinesis_firehose_response():
         # if data was delivered as json; caches loaded value
         data = record.data_as_text
 
+        metadata_partition = KinesisFirehoseResponseRecordMetadata(partition_keys={"year": 2023})
         processed_record = KinesisFirehoseResponseRecord(
             record_id=record.record_id,
             result="Ok",
+            metadata=metadata_partition,
         )
         processed_record.data_from_text(data=data)
         response.add_record(record=processed_record)
-    response_dict = response.asdict
+    response_dict = response.asdict()
 
     res_records = list(response_dict["records"])
     assert len(res_records) == 2
@@ -30,6 +33,7 @@ def test_kinesis_firehose_response():
     assert record_01["result"] == "Ok"
     assert record_01["recordId"] == record01_raw["recordId"]
     assert record_01["data"] == record01_raw["data"]
+    assert record_01["metadata"]["partitionKeys"]["year"] == 2023
 
     assert response.records[0].data_as_bytes == b"Hello World"
     assert response.records[0].data_as_text == "Hello World"
@@ -49,7 +53,7 @@ def test_kinesis_firehose_create_response():
         )
         processed_record.data_from_text(data=data)
         response.add_record(record=processed_record)
-    response_dict = response.asdict
+    response_dict = response.asdict()
 
     res_records = list(response_dict["records"])
     assert len(res_records) == 2