fix: allow passing list or tuple for methods to the route function for ApiGatewayResolver

Author: Tom McCarthy

aws_lambda_powertools/event_handler/api_gateway.py

@@ -10,7 +10,7 @@
 from enum import Enum
 from functools import partial
 from http import HTTPStatus
-from typing import Any, Callable, Dict, List, Optional, Set, Union
+from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union
 
 from aws_lambda_powertools.event_handler import content_types
 from aws_lambda_powertools.event_handler.exceptions import ServiceError
@@ -453,27 +453,30 @@ def __init__(
     def route(
         self,
         rule: str,
-        method: str,
+        method: Union[str, Union[List[str], Tuple[str]]],
         cors: Optional[bool] = None,
         compress: bool = False,
         cache_control: Optional[str] = None,
     ):
         """Route decorator includes parameter `method`"""
 
         def register_resolver(func: Callable):
-            logger.debug(f"Adding route using rule {rule} and method {method.upper()}")
+            methods = (method,) if isinstance(method, str) else method
+            logger.debug(f"Adding route using rule {rule} and methods: {','.join((m.upper() for m in methods))}")
             if cors is None:
                 cors_enabled = self._cors_enabled
             else:
                 cors_enabled = cors
-            self._routes.append(Route(method, self._compile_regex(rule), func, cors_enabled, compress, cache_control))
-            route_key = method + rule
-            if route_key in self._route_keys:
-                warnings.warn(f"A route like this was already registered. method: '{method}' rule: '{rule}'")
-            self._route_keys.append(route_key)
-            if cors_enabled:
-                logger.debug(f"Registering method {method.upper()} to Allow Methods in CORS")
-                self._cors_methods.add(method.upper())
+
+            for item in methods:
+                self._routes.append(Route(item, self._compile_regex(rule), func, cors_enabled, compress, cache_control))
+                route_key = item + rule
+                if route_key in self._route_keys:
+                    warnings.warn(f"A route like this was already registered. method: '{item}' rule: '{rule}'")
+                self._route_keys.append(route_key)
+                if cors_enabled:
+                    logger.debug(f"Registering method {item.upper()} to Allow Methods in CORS")
+                    self._cors_methods.add(item.upper())
             return func
 
         return register_resolver
@@ -679,14 +682,13 @@ def __init__(self):
     def route(
         self,
         rule: str,
-        method: Union[str, List[str]],
+        method: Union[str, Union[List[str], Tuple[str]]],
         cors: Optional[bool] = None,
         compress: bool = False,
         cache_control: Optional[str] = None,
     ):
         def register_route(func: Callable):
-            methods = method if isinstance(method, list) else [method]
-            for item in methods:
-                self._routes[(rule, item, cors, compress, cache_control)] = func
+            methods = (method,) if isinstance(method, str) else tuple(method)
+            self._routes[(rule, methods, cors, compress, cache_control)] = func
 
         return register_route

tests/functional/event_handler/test_api_gateway.py

@@ -1021,3 +1021,39 @@ def get_func_another_duplicate():
     # THEN only execute the first registered route
     # AND print warnings
     assert result["statusCode"] == 200
+
+
+def test_route_multiple_methods():
+    # GIVEN a function with http methods passed as a list
+    app = ApiGatewayResolver()
+    req = "foo"
+    get_event = deepcopy(LOAD_GW_EVENT)
+    get_event["resource"] = "/accounts/{account_id}"
+    get_event["path"] = f"/accounts/{req}"
+
+    post_event = deepcopy(get_event)
+    post_event["httpMethod"] = "POST"
+
+    put_event = deepcopy(get_event)
+    put_event["httpMethod"] = "PUT"
+
+    lambda_context = {}
+
+    @app.route(rule="/accounts/<account_id>", method=["GET", "POST"])
+    def foo(account_id):
+        assert app.lambda_context == lambda_context
+        assert account_id == f"{req}"
+        return {}
+
+    # WHEN calling the event handler with the supplied methods
+    get_result = app(get_event, lambda_context)
+    post_result = app(post_event, lambda_context)
+    put_result = app(put_event, lambda_context)
+
+    # THEN events are processed correctly
+    assert get_result["statusCode"] == 200
+    assert get_result["headers"]["Content-Type"] == content_types.APPLICATION_JSON
+    assert post_result["statusCode"] == 200
+    assert post_result["headers"]["Content-Type"] == content_types.APPLICATION_JSON
+    assert put_result["statusCode"] == 404
+    assert put_result["headers"]["Content-Type"] == content_types.APPLICATION_JSON