Feature request: Allow the ApiGatewayResolver to support an existing subclass of BaseProxyEvent

[BVMiko]

Use case

I often want to retrieve values from a ProxyEvent using the helper functions prior to the app.resolve() call; and find myself creating a second instance for it, for example:

@event_source(data_class=APIGatewayProxyEvent)
def lambda_handler(event: APIGatewayProxyEvent, context: LambdaContext):
    logger.append_keys(user_agent=event.get_header_value("user-agent", "[No User-Agent]"))
    return app.resolve(event.raw_event, context)

Where I feel I should just be able to pass in the instance which was already created:

    return app.resolve(event, context)

I've now found myself wanting to use the ApiGatewayResolver class with a Lambda triggered by EventBridge Rules, and found that it can be done trivially with a custom ProxyEvent subclass -- but only if the ApiGatewayResolver accepts the custom ProxyEvent.

The changes required to support a custom ProxyEvent are very minimal and shouldn't be backwards-incompatible.

I see there was an adjustment about 18 months ago in #1152 & #1159 where a warning message and fallback were added; my proposed change involves replacing the warning with fully supporting the existing ProxyEvent class.

Solution/User Experience

@event_source(data_class=APIGatewayProxyEvent)
def lambda_handler(event: APIGatewayProxyEvent, context: LambdaContext):
    logger.append_keys(user_agent=event.get_header_value("user-agent", "[No User-Agent]"))
    return app.resolve(event.raw_event, context)

becomes

@event_source(data_class=APIGatewayProxyEvent)
def lambda_handler(event: APIGatewayProxyEvent, context: LambdaContext):
    logger.append_keys(user_agent=event.get_header_value("user-agent", "[No User-Agent]"))
    return app.resolve(event, context)

Also it allows a custom ProxyEvent to be used such as:

class CustomProxyEvent(BaseProxyEvent):
    def __init__(self, data: dict[str, Any], json_deserializer: Optional[Callable]=None):
        data_wrapper = {"path": "", "httpMethod": data["action"], "body": json.dumps(data)}
        super().__init__(data_wrapper, json_deserializer)
        self._json_data = data

    @property
    def action(self) -> str:
        return cast(str, cast(dict, self._json_data)["action"])

    @property
    def content(self) -> str:
        return cast(str, cast(dict, self._json_data)["content"])

    # Unfortunately this must still be provided unless we mark it as optional.
    # I have a separate commit prepared, if you are willing to consider I can include it
    def header_serializer(self) -> BaseHeadersSerializer:
        return MultiValueHeadersSerializer()

app = ApiGatewayResolver(debug=True)
logger = Logger()

@app.route("", "foo")
def group_stats_get() -> Response:
    return Response(200, body=app.current_event.content)

@event_source(data_class=CustomProxyEvent)
def handler(event: CustomProxyEvent, context: LambdaContext):
    logger.append_keys(action=event.action)
    return app.resolve(event, context)

with the data

{"action": "foo", "content": "bar"}

Alternative solutions

No response

Acknowledgment

• This feature request meets Powertools for AWS Lambda (Python) Tenets
• Should this be considered in other Powertools for AWS Lambda languages? i.e. Java, TypeScript, and .NET

[leandrodamascena]

Hey @BVMiko! Thanks for opening this issue and send the PR! We will review this week.

[heitorlessa]

Adding this back as Triage as we got sidetracked with re:Invent and operational issues

[BVMiko]

Hey, I just wanted to ping and see if anyone had chance to check this issue and corresponding PR yet. The current code's limitation doesn't seem to be necessary; and this seems like a pretty safe enhancement to accept any subclass of a proxy event.

I've been using a fork with this commit for several projects over the last few months and it's been handy for several reasons:

• The ability to combine the @event_source() decorator with the ApiGatewayResolver
• The ability to use a custom proxy event for input processing

It would also make sense to make the header_serializer() an optional method on subclasses of BaseProxyEvent, any resulting proxy event class implementations could be simplified.

[heitorlessa]

I replied in the PR back then but didn't copy it here, sorry, pasting verbatim.

---

hey @BVMiko apologies on behalf of the team, we dropped the ball in coming back with a timely counter-offer for this extension.

How about we work on a RFC to make it possible to route anything with a generic Event Handler?

While this applies to you, we already had customers telling us in private that they'd love an Event Handler for EventBridge, S3, Step Functions, SNS, etc.

It'd be great to have a way to (1) specify which key(s) of the event to do the routing, and (2) declare the shape of the event available in current_event.

That way, we can have a dedicated documentation for this, and future-proof new Event Handler specializations as it gets popular, just like we have REST and GraphQL Event Handlers today.

What do you say?