feat(batch): new BatchProcessor for SQS, DynamoDB, Kinesis

[heitorlessa]

Issue #, if available: #858

Description of changes:

Implements new Batch Processor following the new built-in partial response handling for Lambda SQS, DynamoDB Stream, and Kinesis Data Streams.

Checklist

• Meet tenets criteria
• Update tests
• Update docs
• PR title follows conventional commit semantics
• Add DynamoDB Stream support
• Add Kinesis Data Stream support
• Think of a better name other than .report() -> response to match Java counterpart
• Support event source data classes by default
• Mypy support
• Review with @cakepietoast whether we still need suppress_exception behaviour
• Bring-your-own-model Pydantic Parser as an option
• Test for incorrect data/payload that won't be parsed (either Model or Data Class)
• Raise BatchProcessingError when all batch records fail to ensure failure arises

Tasks to be created in separate PR to ease reviewing

• Support for handling permanent exceptions (might be for the release after next)

UX

As a decorator

from aws_lambda_powertools.utilities.batch import batch_processor, BatchProcessor, EventType
from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord

processor = BatchProcessor(event_type=EventType.SQS) # or, EventType.KinesisDataStreams, EventType.DynamoDB

def record_handler(record: SQSRecord): # or DynamoDBRecord, KinesisStreamRecord
    return "success"

@batch_processor(record_handler=record_handler, processor=processor)
def lambda_handler(event, context):
    return processor.response() # this is necessary to ensure Lambda returns the correct response

As a context manager, in case you need full access to the batch processed, or handle exceptions yourself

from aws_lambda_powertools.utilities.batch import batch_processor, BatchProcessor, EventType

processor = BatchProcessor(event_type=EventType.SQS) 


def lambda_handler(event, context):
    records = event["Records"]
    with processor(records, record_handler) as batch:
        processed_messages = batch.process()
    
    return batch.response() # this is necessary to ensure Lambda returns the correct response

Breaking change checklist

RFC issue #:

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

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

[codecov-commenter]

Codecov Report

Merging #886 (95715d0) into develop (99227ce) will decrease coverage by 0.09%.
The diff coverage is 97.24%.

@@             Coverage Diff             @@
##           develop     #886      +/-   ##
===========================================
- Coverage    99.88%   99.79%   -0.10%     
===========================================
  Files          118      118              
  Lines         5161     5262     +101     
  Branches       578      596      +18     
===========================================
+ Hits          5155     5251      +96     
- Misses           2        6       +4     
- Partials         4        5       +1     

Impacted Files	Coverage Δ
aws_lambda_powertools/utilities/batch/base.py	97.67% <96.70%> (-2.33%)	⬇️
aws_lambda_powertools/utilities/batch/__init__.py	100.00% <100.00%> (ø)
...ws_lambda_powertools/utilities/batch/exceptions.py	100.00% <100.00%> (ø)
aws_lambda_powertools/utilities/parser/parser.py	93.93% <0.00%> (-6.07%)	⬇️

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 99227ce...95715d0. Read the comment docs.

[heitorlessa]

@cakepietoast ready to review implementation before it gets bigger with mypy and doc stuff.

[ran-isenberg]

@heitorlessa as discussed, here's the sqs example:

@validator('body', pre=True)
def transform_body_to_dict(cls, value):
        return json.loads(value)

[heitorlessa]

🤦‍♂️ yes! Not sure why I used Callback

…

On Wed, 15 Dec 2021 at 09:54, Tom McCarthy ***@***.***> wrote: ***@***.**** commented on this pull request. ------------------------------ In aws_lambda_powertools/utilities/batch/base.py <#886 (comment)> : > @@ -72,7 +110,7 @@ def __call__(self, records: List[Any], handler: Callable): self.handler = handler return self - def success_handler(self, record: Any, result: Any): + def success_handler(self, record, result: Any) -> SuccessCallback: Would SuccessResponse be a more accurate name? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#886 (review)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAZPQBFPYJNIEG5KWW7JBGDURBJTJANCNFSM5JZVI3AA> .

[heitorlessa]

Great point. It’s good to anticipate someone will subclass it sooner or later

…

On Tue, 14 Dec 2021 at 15:54, Guilherme Martins Crocetti < ***@***.***> wrote: ***@***.**** commented on this pull request. ------------------------------ In aws_lambda_powertools/utilities/batch/base.py <#886 (comment)> : > +class BatchProcessor(BasePartialProcessor): + DEFAULT_RESPONSE: Dict[str, List[Optional[dict]]] = {"batchItemFailures": []} + + def __init__(self, event_type: EventType, model: Optional["BatchTypeModels"] = None): + """Process batch and partially report failed items + + Parameters + ---------- + event_type: EventType + Whether this is a SQS, DynamoDB Streams, or Kinesis Data Stream event + model: Optional["BatchTypeModels"] + Parser's data model using either SqsRecordModel, DynamoDBStreamRecordModel, KinesisDataStreamRecord + """ + self.event_type = event_type + self.model = model + self.batch_response = self.DEFAULT_RESPONSE Hey @heitorlessa <https://github.com/heitorlessa> , don't know if this class is meant to be extended. Case it's not, following comment won't make much sense but anyways: Do you think it's worth being more defensive in here by adding a deepcopy ? ⬇️ Suggested change - self.batch_response = self.DEFAULT_RESPONSE + self.batch_response = deepcopy(self.DEFAULT_RESPONSE) Just to avoid a scenario where someone extends from this class and mess with the DEFAULT_RESPONSE through self.batch_response (self.batch_response.update({"batchItemFailures": []}) for instance ) - or any sort of changes... — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#886 (review)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAZPQBDWTSDMHXO55IOJODDUQ5LBTANCNFSM5JZVI3AA> .

[jplock]

would it make sense to have separate processors for each event type (SQS, DynamoDB or Kinesis) instead of growing the complexity of this class? Then you could encapsulate the failure collection in the specific processor.

[heitorlessa]

That was the initial version we wanted to implemented - a KinesisDataStreamProcessor, DynamoDB... then @cakepietoast argued that this was gonna confuse customers with other available processors (Sqs, PartialProcessor, BaseProcessor), as we can only deprecate them in v2.

I'm 50/50 here if I'm honest. I prefer a separate one but I also can see customers easily confused of which one to pick despite docs change I'm gonna make.

Implementation wise, this will remain stable. The only two changes I can anticipate is 1/ supporting the new Permanent Exception parameter, and 2/ raising a descriptive exception in case we reach an AttributeError when collecting message id/sequence number from a malformed event/model.

[jplock]

What if we changed the nomenclature to be “producer” and “consumer” (these processors would be consumers). I had that other idea earlier to make it easier to use the SQS and DynamoDB batch write methods taking into account their batch sizes, those could be “producers” 🤷‍♂️

[heitorlessa]

Where would the producer sit and what would be its responsibilities?

For that suggestion on partitioning, we should add it to the Event Source Data Class as it's a no brainier.

I think the word Consumer wouldn't be explicit enough on the capabilities Batch provide - maybe something else then?

Features

• Transform incoming records into Event Source Data Class or Pydantic Models
• Call an user defined function for each record in the batch
• Keep track of exceptions raised for any record
• For partial successes, extract message IDs (SQS) or sequence numbers (DynamoDB & Kinesis) and build a response to the new BatchItemIdentifiers contract
• Future: If a raised exception matches the permanent_exceptions set, send these records to the configured DLQ in batches of 10

[heitorlessa]

Docs finally done - @michaelbrewer @ran-isenberg could you have a quick pass before we release it tomorrow? All I need to do now is merge and start writing the release notes.

Here's what I've changed:

• Updated the docs to reflect the new BatchProcessor in favour of sqs_batch_processor and PartialSQSProcessor
• Added a Migration guide for those using sqs_batch_processor and PartialSQSProcessor
• Added Caveats section in case people use Tracer response auto-capture feature with this utility in large batch sizes
• Added sample event and sample response for each
• Added SAM infrastructure for SQS, Kinesis and DynamoDB along with DLQ and IAM Permissions
• Added an Extensibility section
• Added Pydantic section
• Updated Create your own processor
• Updated Key features
• Updated Intro
• Recreated Background
• Repurposed a few previous sections like Sentry integration under Legacy

[michaelbrewer]

@heitorlessa super cool. i will try to play around with it once merged

[ran-isenberg]

@heitorlessa looks great, thank you very much for putting the effort supporting both parser and data classes.