Support for S3 Events

[djmango]

Coming from Chalice, wrote a rudementary custom LambdaHandler for S3 events. For this I'm just creating a router that handles all /aws routes, in this case /aws/s3. Lots of improvements to make before I'd open a PR to merge this feature, but unsure of next steps. The class as of today is below.

from mangum.types import LambdaEvent, LambdaContext, LambdaConfig, Response, Scope, LambdaHandler
from mangum.handlers.utils import maybe_encode_body, handle_multi_value_headers, handle_exclude_headers, handle_base64_response_body


class S3EventHandler(LambdaHandler):
    """ Handles S3 events for Mangum. """

    @classmethod
    def infer(cls, event: LambdaEvent, context: LambdaContext, config: LambdaConfig) -> bool:
        """ Returns True if the event is a S3 event. """
        return "Records" in event and len(event["Records"]) == 1 and "eventSource" in event["Records"][0] and event["Records"][0]["eventSource"] == "aws:s3"

    def __init__(self, event: LambdaEvent, context: LambdaContext, config: LambdaConfig) -> None:
        self.event = event
        self.context = context
        self.config = config

    @property
    def body(self) -> bytes:
        return maybe_encode_body(self.event.get("body", b""), is_base64=self.event.get("isBase64Encoded", False),)

    @property
    def scope(self) -> Scope:
        scope: Scope = {
            "type": "http",
            "method": 'PUT' if self.event["Records"][0]["eventName"] == "ObjectCreated:Put" else 'DELETE',
            "http_version": "1.1",
            "headers": {},  # https://github.com/khornberg/cloud-events-handler/blob/master/src/cloud_events/adapter.py
            "path": '/aws/s3',
            "raw_path": None,
            "root_path": "",
            "scheme": "https",
            "query_string": "",
            "client": "",
            "asgi": {"version": "3.0", "spec_version": "2.0"},
            "aws.event": self.event,
            "aws.context": self.context,
        }
        return scope

    def __call__(self, response: Response) -> dict:
        finalized_headers, multi_value_headers = handle_multi_value_headers(response["headers"])
        finalized_body, is_base64_encoded = handle_base64_response_body(response["body"], finalized_headers, self.config["text_mime_types"])

        return {
            "statusCode": response["status"],
            "headers": handle_exclude_headers(finalized_headers, self.config),
            "multiValueHeaders": handle_exclude_headers(
                multi_value_headers, self.config
            ),
            "body": finalized_body,
            "isBase64Encoded": is_base64_encoded,
        }

[jordaneremieff]

Hi @djmango, I'll have to think about this a bit more. I've not actually considered many potential handler classes might end up in this repo and what might make more sense to support externally. This should probably be fine, I'll try to find some time to look into it soon.

[lorenabalan]

Hey @jordaneremieff I'm happy to raise a docs-related PR to add an example for how one can use custom handlers. I only managed to get something like that working by coming across this issue for inspiration, and by digging through the source code. I reckon it'd be valuable for others if they have a more straightforward guide in the docs.
Thanks for an awesome library btw. 🙂

[inkhey]

i really like this idea @lorenabalan.
Can i suggest something for the future ?
It may be nice in the future to not have to mock events like this is http. i discover https://www.asyncapi.com/ and https://github.com/Lancetnik/Propan and think there is probably something to do with it to have documentation for non-http events.