Convert from callable to use route; add test; fix up formatting

Author: Brian Villemarette

aws_lambda_powertools/event_handler/api_gateway.py

@@ -629,41 +629,57 @@ def _to_response(self, result: Union[Dict, Response]) -> Response:
     def _json_dump(self, obj: Any) -> str:
         return self._serializer(obj)
 
-class Blueprint():
+
+class Blueprint:
     """Blueprint helper class to allow splitting ApiGatewayResolver into multiple files"""
+
     def __init__(self):
         self._api: Dict[tuple, Callable] = {}
 
-    def __call__(self, rule: str, method: Union[str, Tuple[str], List[str]], cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None,):
+    def route(
+        self,
+        rule: str,
+        method: Union[str, Tuple[str], List[str]],
+        cors: Optional[bool] = None,
+        compress: bool = False,
+        cache_control: Optional[str] = None,
+    ):
         def actual_decorator(func: Callable):
             @wraps(func)
             def wrapper(app: ApiGatewayResolver):
                 def inner_wrapper():
                     return func(app)
+
                 return inner_wrapper
+
             if isinstance(method, (list, tuple)):
                 for item in method:
                     self._api[(rule, item, cors, compress, cache_control)] = wrapper
             else:
                 self._api[(rule, method, cors, compress, cache_control)] = wrapper
+
         return actual_decorator
 
     def get(self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None):
-        return self.__call__(rule, "GET", cors, compress, cache_control)
+        return self.route(rule, "GET", cors, compress, cache_control)
 
     def post(self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None):
-        return self.__call__(rule, "POST", cors, compress, cache_control)
+        return self.route(rule, "POST", cors, compress, cache_control)
 
     def put(self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None):
-        return self.__call__(rule, "PUT", cors, compress, cache_control)
+        return self.route(rule, "PUT", cors, compress, cache_control)
 
-    def delete(self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None):
-        return self.__call__(rule, "DELETE", cors, compress, cache_control)
+    def delete(
+        self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None
+    ):
+        return self.route(rule, "DELETE", cors, compress, cache_control)
 
-    def patch(self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None):
-        return self.__call__(rule, "PATCH", cors, compress, cache_control)
+    def patch(
+        self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None
+    ):
+        return self.route(rule, "PATCH", cors, compress, cache_control)
 
-    def register_to_app(self, app:ApiGatewayResolver):
+    def register_to_app(self, app: ApiGatewayResolver):
         """Bind a blueprint object to an existing ApiGatewayResolver instance"""
         for route, func in self._api.items():
             app.route(*route)(func(app=app))

tests/functional/event_handler/test_api_gateway.py

@@ -13,6 +13,7 @@
 from aws_lambda_powertools.event_handler import content_types
 from aws_lambda_powertools.event_handler.api_gateway import (
     ApiGatewayResolver,
+    Blueprint,
     CORSConfig,
     ProxyEventType,
     Response,
@@ -842,3 +843,21 @@ def foo():
     # THEN process event correctly
     assert result["statusCode"] == 200
     assert result["headers"]["Content-Type"] == content_types.APPLICATION_JSON
+
+
+def test_api_gateway_app_proxy():
+    # GIVEN a Blueprint with registered routes
+    app = ApiGatewayResolver()
+    blueprint = Blueprint()
+
+    @blueprint.get("/my/path")
+    def foo(app):
+        return {}
+
+    blueprint.register_to_app(app)
+    # WHEN calling the event handler after applying routes from blueprint object
+    result = app(LOAD_GW_EVENT, {})
+
+    # THEN process event correctly
+    assert result["statusCode"] == 200
+    assert result["headers"]["Content-Type"] == content_types.APPLICATION_JSON