feat(event_sources): extract CloudWatch Logs in Kinesis streams

[heitorlessa]

Issue number: #1488

Summary

Changes

Please provide a summary of what's being changed

This PR introduces two functions to decode, decompress, and extract CloudWatchLogsDecodedData from Kinesis Data Streams.

• extract_cloudwatch_logs_from_event. Returns a list of CloudWatchLogsDecodedData from KinesisStreamEvent. Useful for simple processing.
• extract_cloudwatch_logs_from_record. Returns CloudWatchLogsDecodedData from KinesisStreamRecord. Useful for robust batch processing and seamless integration with Batch Processing

User experience

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

Batch processing

from aws_lambda_powertools.utilities.batch import (BatchProcessor, EventType,
                                                   batch_processor)
from aws_lambda_powertools.utilities.data_classes.kinesis_stream_event import (
    KinesisStreamRecord, extract_cloudwatch_logs_from_record)

processor = BatchProcessor(event_type=EventType.KinesisDataStreams)


def record_handler(record: KinesisStreamRecord):
    log = extract_cloudwatch_logs_from_record(record)
    return log.message_type == "DATA_MESSAGE"


@batch_processor(record_handler=record_handler, processor=processor)
def lambda_handler(event, context):
    return processor.response()

Simple processing

from typing import List

from aws_lambda_powertools.utilities.data_classes import event_source
from aws_lambda_powertools.utilities.data_classes.cloud_watch_logs_event import \
    CloudWatchLogsDecodedData
from aws_lambda_powertools.utilities.data_classes.kinesis_stream_event import (
    KinesisStreamEvent, extract_cloudwatch_logs_from_event)


@event_source(data_class=KinesisStreamEvent)
def simple_handler(event: KinesisStreamEvent, context):
    logs: List[CloudWatchLogsDecodedData] = extract_cloudwatch_logs_from_event(event)
    for log in logs:
        if log.message_type == "DATA_MESSAGE":
            return "success"
    return "success"

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.

---

View rendered docs/utilities/data_classes.md

[heitorlessa]

customer approved design. Merging after extra due diligence since @rubenfonseca and @leandrodamascena are unavailable.