feat(event-handler): new resolvers to fix current_event typing

aws_lambda_powertools/event_handler/api_gateway.py

@@ -731,3 +731,84 @@ def register_route(func: Callable):
             self._routes[(rule, methods, cors, compress, cache_control)] = func
 
         return register_route
+
+
+class APIGatewayProxyEventResolver(ApiGatewayResolver):
+    current_event: APIGatewayProxyEvent
+
+    def __init__(
+        self,
+        cors: Optional[CORSConfig] = None,
+        debug: Optional[bool] = None,
+        serializer: Optional[Callable[[Dict], str]] = None,
+        strip_prefixes: Optional[List[str]] = None,
+    ):
+        """
+        Parameters
+        ----------
+        cors: CORSConfig
+            Optionally configure and enabled CORS. Not each route will need to have to cors=True
+        debug: Optional[bool]
+            Enables debug mode, by default False. Can be also be enabled by "POWERTOOLS_EVENT_HANDLER_DEBUG"
+            environment variable
+        serializer : Callable, optional
+            function to serialize `obj` to a JSON formatted `str`, by default json.dumps
+        strip_prefixes: List[str], optional
+            list of prefixes to be removed from the request path before doing the routing. This is often used
+            with api gateways with multiple custom mappings.
+        """
+        super().__init__(ProxyEventType.APIGatewayProxyEvent, cors, debug, serializer, strip_prefixes)
+
+
+class APIGatewayProxyEventV2Resolver(ApiGatewayResolver):
+    current_event: APIGatewayProxyEventV2
+
+    def __init__(
+        self,
+        cors: Optional[CORSConfig] = None,
+        debug: Optional[bool] = None,
+        serializer: Optional[Callable[[Dict], str]] = None,
+        strip_prefixes: Optional[List[str]] = None,
+    ):
+        """
+        Parameters
+        ----------
+        cors: CORSConfig
+            Optionally configure and enabled CORS. Not each route will need to have to cors=True
+        debug: Optional[bool]
+            Enables debug mode, by default False. Can be also be enabled by "POWERTOOLS_EVENT_HANDLER_DEBUG"
+            environment variable
+        serializer : Callable, optional
+            function to serialize `obj` to a JSON formatted `str`, by default json.dumps
+        strip_prefixes: List[str], optional
+            list of prefixes to be removed from the request path before doing the routing. This is often used
+            with api gateways with multiple custom mappings.
+        """
+        super().__init__(ProxyEventType.APIGatewayProxyEventV2, cors, debug, serializer, strip_prefixes)
+
+
+class ALBEventResolver(ApiGatewayResolver):
+    current_event: ALBEvent
+
+    def __init__(
+        self,
+        cors: Optional[CORSConfig] = None,
+        debug: Optional[bool] = None,
+        serializer: Optional[Callable[[Dict], str]] = None,
+        strip_prefixes: Optional[List[str]] = None,
+    ):
+        """
+        Parameters
+        ----------
+        cors: CORSConfig
+            Optionally configure and enabled CORS. Not each route will need to have to cors=True
+        debug: Optional[bool]
+            Enables debug mode, by default False. Can be also be enabled by "POWERTOOLS_EVENT_HANDLER_DEBUG"
+            environment variable
+        serializer : Callable, optional
+            function to serialize `obj` to a JSON formatted `str`, by default json.dumps
+        strip_prefixes: List[str], optional
+            list of prefixes to be removed from the request path before doing the routing. This is often used
+            with api gateways with multiple custom mappings.
+        """
+        super().__init__(ProxyEventType.ALBEvent, cors, debug, serializer, strip_prefixes)

tests/functional/event_handler/test_api_gateway.py

@@ -12,6 +12,9 @@
 
 from aws_lambda_powertools.event_handler import content_types
 from aws_lambda_powertools.event_handler.api_gateway import (
+    ALBEventResolver,
+    APIGatewayProxyEventResolver,
+    APIGatewayProxyEventV2Resolver,
     ApiGatewayResolver,
     CORSConfig,
     ProxyEventType,
@@ -47,13 +50,14 @@ def read_media(file_name: str) -> bytes:
 
 
 def test_alb_event():
-    # GIVEN a Application Load Balancer proxy type event
-    app = ApiGatewayResolver(proxy_type=ProxyEventType.ALBEvent)
+    # GIVEN an Application Load Balancer proxy type event
+    app = ALBEventResolver()
 
     @app.get("/lambda")
     def foo():
         assert isinstance(app.current_event, ALBEvent)
         assert app.lambda_context == {}
+        assert app.current_event.request_context.elb_target_group_arn is not None
         return Response(200, content_types.TEXT_HTML, "foo")
 
     # WHEN calling the event handler
@@ -68,12 +72,13 @@ def foo():
 
 def test_api_gateway_v1():
     # GIVEN a Http API V1 proxy type event
-    app = ApiGatewayResolver(proxy_type=ProxyEventType.APIGatewayProxyEvent)
+    app = APIGatewayProxyEventResolver()
 
     @app.get("/my/path")
     def get_lambda() -> Response:
         assert isinstance(app.current_event, APIGatewayProxyEvent)
         assert app.lambda_context == {}
+        assert app.current_event.request_context.domain_name == "id.execute-api.us-east-1.amazonaws.com"
         return Response(200, content_types.APPLICATION_JSON, json.dumps({"foo": "value"}))
 
     # WHEN calling the event handler
@@ -106,12 +111,13 @@ def get_lambda() -> Response:
 
 def test_api_gateway_v2():
     # GIVEN a Http API V2 proxy type event
-    app = ApiGatewayResolver(proxy_type=ProxyEventType.APIGatewayProxyEventV2)
+    app = APIGatewayProxyEventV2Resolver()
 
     @app.post("/my/path")
     def my_path() -> Response:
         assert isinstance(app.current_event, APIGatewayProxyEventV2)
         post_data = app.current_event.json_body
+        assert app.current_event.cookies[0] == "cookie1"
         return Response(200, content_types.TEXT_PLAIN, post_data["username"])
 
     # WHEN calling the event handler