Feature request: Support S3 Event Notifications Model for Event Bridge

[robert-malai]

Use case

With the introduction of S3 Event Notifications in Event Bridge here we need the ability to quickly unpack the event payload from the Event Bridge Envelope.

The structure of the message is described here.

Solution/User Experience

It would suffice to extend the s3 models in the the parser module.

Alternative solutions

Have the module instantiated at the project level and just use the envelope unpacking mechanism.

Acknowledgment

• This feature request meets Lambda Powertools Tenets
• Should this be considered in other Lambda Powertools languages? i.e. Java, TypeScript

[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 AWS Lambda Powertools Discord: Invite link

[heitorlessa]

@robert-malai thank you for the feature request! Besides the data class, are there any repetitive task that we could make it easier as part of this new data class?

Added the correct labels to watch for +1 to help us prioritize. We'd welcome a PR to fast track this too!

Thanks a lot!

[theipster]

On a similar theme, we have a use-case for unpacking SQS-wrapped S3 event notifications, which would be nice to pull off the shelf.

(Apologies if there's an existing way to achieve this, but I couldn't see anything in the docs.)

A really simple additional model seems to do the trick:

from aws_lambda_powertools.utilities.parser.models import SqsRecordModel, S3Model
from pydantic import Json

class SqsS3EventNotificationModel(SqsRecordModel):
  body: Json[S3Model]

This could then be used as just another model, e.g.:

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

processor = BatchProcessor(event_type=EventType.SQS, model=SqsS3EventNotificationModel)

def record_handler(record: SqsS3EventNotificationModel):
  for s3_record in record.body.Records:
    do_something_useful(s3_record)

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

P.S. if there's a way to skip the intermediate step, that would be even nicer. In other words, rather than needing to define a def record_handler(record: SqsS3EventNotificationModel) that iterates through the record.body.Records, it would be nice to just directly define a def record_handler(record: S3Model) and let the processor unwrap everything instead. This would keep the business logic clean, and not having to care whether it came via SQS or EventBridge, etc..

[robert-malai]

@heitorlessa
Sorry for the late response.

Not sure which other repetitive tasks would benefit out of this dataclass. Seems to be pretty specific to the EventBridge notification for S3 events.

Why would someone use EventBridge events for an S3 Bucket? Well ... you can fan out events w/o using an SNS, for simple, non-extensible setups.

[heitorlessa]

Gotcha! I assumed you were also thinking on going beyond Parser use case, so we can enrich EventBridge experience.

If anyone got the bandwidth, we'd welcome any PR to add a new model class for this feature request.

@theipster for the Json[Model], could you create a feature request on this please? We'd need to extend typing to allow Json[Model] otherwise Mypy/Pyright will fail for most customers. Agree it's a lot simpler, we just didn't know at the time about Json in Pydantic and its performance implications - thank you for bringing it up!

[ivica-k]

I'll try to implement this one.

[heitorlessa]

thank you @ivica-k - Please reach out if you need any help whatsoever while you're at it. We have both #python and #contributing channel on Discord