feat(event_source): add Kinesis Firehose Data Transformation data class

[roger-zhangg]

Issue number:#2440

Summary

Support Kinesis Firehose Response Record data class in event_sources.

This PR is still in draft status, much to discuss including

• Dataclasses are mostly used for read only Dicts, this request need to construct dict before using the data classes, which is different. Need to review on the current approach
• Discuss what functionality should be supported. E.g (data.from_str, data.from_binary)

Changes

Please provide a summary of what's being changed
Added three data_classes and their constructor that supports Kinesis Firehose Response Record

User experience

Please share what the user experience looks like before and after this change

BEFORE

import base64

print('Loading function')


def lambda_handler(event, context):
    output = []

    for record in event['records']:
        print(record['recordId'])
        payload = base64.b64decode(record['data']).decode('utf-8')

        # Do custom processing on the payload here

        output_record = {
            'recordId': record['recordId'],
            'result': 'Ok',
            'data': base64.b64encode(payload.encode('utf-8')).decode('utf-8')
        }
        output.append(output_record)

    print('Successfully processed {} records.'.format(len(event['records'])))

    return {'records': output}

AFTER

import base64

from aws_lambda_powertools.utilities.data_classes import (
    KinesisFirehoseDataTransformationRecord,
    KinesisFirehoseDataTransformationResponse,
)
from aws_lambda_powertools.utilities.serialization import base64_from_json
from aws_lambda_powertools.utilities.typing import LambdaContext


def lambda_handler(event: dict, context: LambdaContext):
    result = KinesisFirehoseDataTransformationResponse()

    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
        transformed_data = {"tool_used": "powertools_dataclass", "original_payload": payload}

        processed_record = KinesisFirehoseDataTransformationRecord(
            record_id=record["recordId"],
            data=base64_from_json(transformed_data),
        )
        result.add_record(processed_record)

    # return transformed records
    return result.asdict()

Checklist

If your change doesn't seem to apply, please leave them unchecked.

• Meet tenets criteria
• I have performed a self-review of this change
• Changes have been tested
• Changes are documented
• PR title follows conventional commit semantics

Is this a breaking change?

RFC issue number:

Checklist:

• Migration process documented
• Implement warnings (if it can live side by side)

Acknowledgment

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

Disclaimer: We value your time and bandwidth. As such, any pull requests created on non-triaged issues might not be successful.

[roger-zhangg]

from here

powertools-lambda-python/aws_lambda_powertools/utilities/data_classes/common.py

Lines 13 to 14 in 04cf87f

	class DictWrapper(Mapping):
	"""Provides a single read only access to a wrapper dict"""

This DictWrapper is designed to be read-only and used by all data-classes. Currently all data-classes in powertools are used to parse events. But for this issue, we are using data-classes to create the event. And this use case doesn't exist in powertools data-classes yet.
In this case I can't set value directly to data class expect init the class with data parameter. As you can see in my current code,

powertools-lambda-python/aws_lambda_powertools/utilities/data_classes/kinesis_firehose_response.py

Lines 89 to 103 in 6eb8530

	def KinesisFirehoseResponceRecordFactory(
	record_id: str,
	result: Union[FirehoseStateOk, FirehoseStateDropped, FirehoseStateFailed],
	data: str,
	metadata: Optional[KinesisFirehoseResponseRecordMetadata] = None,
	json_deserializer: Optional[Callable] = None,
	) -> KinesisFirehoseResponceRecord:
	pass_data = {
	"recordId": record_id,
	"result": result,
	"data": base64.b64encode(data.encode("utf-8")).decode("utf-8"),
	}
	if metadata:
	data["metadata"] = metadata
	return KinesisFirehoseResponceRecord(data=pass_data, json_deserializer=json_deserializer)

I'm using a factory method to create each of these data classes. But I don't think this is a good solution. So it's better we can discuss to figure out a better way to create these data classes

From my perspective I can think of three possible solution of to support this

• Add a setter method to DictWrapper and init the class as a data class

# possible UX
result = KinesisFirehoseResponce({})
for record in event.records:
        data = record.data_as_json
        processed_record = KinesisFirehoseResponceRecord(
            record_id=record.record_id,
            result=FirehoseStateOk,
        )
        processed_record.data_from_json(data)
        result.add_record(processed_record)
return result

• Add a new Wrapper class to support this specific case for creating events in lambda

# possible UX
        processed_record = KinesisFirehoseResponceRecord(
            record_id=record.record_id,
            result=FirehoseStateOk,
            data=data,
        )

• Do nothing, just keep the current data approach

# possible UX
data={
    "record_id":record.record_id,
    "result":FirehoseStateOk,
    "data":data_raw,
}
record = KinesisFirehoseResponceRecord(data=data)

[leandrodamascena]

Hi @roger-zhangg! I'm reviewing the code, your comments, and reading the AWS documentation for Kinesis Firehose Response, but in the meantime, I think you should fix these issues.

Please mark this PR as "ready to review".

[codecov-commenter]

Codecov Report

Patch coverage: 69.56% and project coverage change: -0.23% ⚠️

Comparison is base (f673368) 96.36% compared to head (6f95e8f) 96.13%.

❗ Your organization needs to install the Codecov GitHub app to enable full functionality.

Additional details and impacted files

@@             Coverage Diff             @@
##           develop    #3029      +/-   ##
===========================================
- Coverage    96.36%   96.13%   -0.23%     
===========================================
  Files          185      186       +1     
  Lines         8064     8131      +67     
  Branches      1509     1525      +16     
===========================================
+ Hits          7771     7817      +46     
- Misses         236      252      +16     
- Partials        57       62       +5     

Files Changed	Coverage Δ
...s/utilities/data_classes/kinesis_firehose_event.py	83.19% <66.66%> (-15.22%)	⬇️
aws_lambda_powertools/utilities/serialization.py	81.81% <81.81%> (ø)
...mbda_powertools/utilities/data_classes/__init__.py	100.00% <100.00%> (ø)

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[leandrodamascena]

• Add a setter method to DictWrapper and init the class as a data class

# possible UX
result = KinesisFirehoseResponce({})
for record in event.records:
        data = record.data_as_json
        processed_record = KinesisFirehoseResponceRecord(
            record_id=record.record_id,
            result=FirehoseStateOk,
        )
        processed_record.data_from_json(data)
        result.add_record(processed_record)
return result

I like this UX, easy and clean to create records and add these in the response. I'm just wondering if we really need DictWrapper. As you said, the DictWrapper class is very much meant for parsing events and getting elements, what do you think about just keeping this class as @dataclasses?

[leandrodamascena]

Please review the class name, there is a typo: Responce instead of Response.

Thanks

[roger-zhangg]

Thanks for the suggestion, I'll refactor to remove the DictWrapper and keep this UX

[heitorlessa]

You can make this UX even nicer by having a method in the Kinesis Record that returns a standalone KinesisFirehoseResponceRecord. That way you don’t need to ask for the record ID since you’re building it from the record already, and can have more fluid options like JSON to get the payload in the expected shape etc. Play around with it later

…

On Wed, 30 Aug 2023 at 23:43, Roger Zhang ***@***.***> wrote: Thanks for the suggestion, I'll refactor to remove the DictWrapper and keep this UX — Reply to this email directly, view it on GitHub <#3029 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAZPQBF56EU6QI6O55BTIVTXX6XYFANCNFSM6AAAAAA4DXDDDM> . You are receiving this because your review was requested.Message ID: ***@***.***>

[leandrodamascena]

I made some few comments.

[heitorlessa]

coming at this later tonight... had to switch to #3091 as it seemed a potential customer issue

[heitorlessa]

Three quick things only and we can merge <3

• It’s missing “Examples” in docstring (for the ones you believe a customer will be mostly exposed to, 1 example suffice)
• During result validation, you’re creating a list with valid strings for every iteration - you can use a class variable (class definition) and use a tuple (fixed array size for memory footprint)
• result="Ok" is the default now, we can remove it to keep it clean. Then in the docs, we can have a tabbed example with the non-ok examples so they know when to use Dropped and ProcessingFailed.

[heitorlessa]

Item 1 and 2 are done.

---

Docstring experience

API reference experience

[heitorlessa]

As soon as the docs contain both the default (ok), Dropped, and ProcessingFailed, along with key parts highlighted (what to import, where standalone function is used, result.add_record, asdict), and any comments or code annotation... it's good to merge ;)

cc @leandrodamascena

Gonna to go to bed (10:33pm)

[sonarcloud]

Kudos, SonarCloud Quality Gate passed!   

0 Bugs
0 Vulnerabilities
0 Security Hotspots
0 Code Smells

No Coverage information
0.5% Duplication

[heitorlessa]

Docs final touches sent

• Use code annotations to provide links, rich formatting, for comments where you can provide more contextual information. Whereas comments would be limited to a single line (cognitive load)
• Always link to the official docs for authorative source
• Provide information upfront to ease customers learning curve in complex topics - in this case, it'd take a while for customers to understand the nuances between the 3 modes (OK, DROP, FAILED)
• Use realistic examples - always put yourself in the customers shoes, specially when they might not be developers. In the exception example, we want to make it true that you could drop an unwanted record :)
• The failed example now uses a code annotation to explain how Kinesis Firehose handles failure. Adding comments would be a distraction, and even if we were to explain it'd be hard to keep it up to date if it changes. Instead, we go straight to the point and add a link for more details.
• Don't be afraid of using newlines and removing multi-line comments. Whenever there's more than one line of comment, maybe there's a better way of phrasing it, or use code annotation.

[heitorlessa]

Now I can go to sleep hahaha. Thank you SO MUCH both of you for the great work here ;)

Merging

[troyswanson]

Thank you all! This is a very exciting feature for Powertools