feat(event-handler): Add PATCH decorator

Author: Michael Brewer

aws_lambda_powertools/event_handler/api_gateway.py

@@ -16,7 +16,7 @@ class ProxyEventType(Enum):
  api_gateway = http_api_v1
 
 
-class RouteEntry:
+class Route:
  def __init__(
  self, method: str, rule: Any, func: Callable, cors: bool, compress: bool, cache_control: Optional[str]
  ):
@@ -34,7 +34,7 @@ class ApiGatewayResolver:
 
  def __init__(self, proxy_type: Enum = ProxyEventType.http_api_v1):
  self._proxy_type = proxy_type
- self._routes: List[RouteEntry] = []
+ self._routes: List[Route] = []
 
  def get(self, rule: str, cors: bool = False, compress: bool = False, cache_control: str = None):
  return self.route(rule, "GET", cors, compress, cache_control)
@@ -48,14 +48,25 @@ def put(self, rule: str, cors: bool = False, compress: bool = False, cache_contr
  def delete(self, rule: str, cors: bool = False, compress: bool = False, cache_control: str = None):
  return self.route(rule, "DELETE", cors, compress, cache_control)
 
+ def patch(self, rule: str, cors: bool = False, compress: bool = False, cache_control: str = None):
+ return self.route(rule, "PATCH", cors, compress, cache_control)
+
  def route(self, rule: str, method: str, cors: bool = False, compress: bool = False, cache_control: str = None):
  def register_resolver(func: Callable):
- self._append(func, rule, method, cors, compress, cache_control)
+ self._add(func, rule, method, cors, compress, cache_control)
  return func
 
  return register_resolver
 
- def resolve(self, event: Dict, context: LambdaContext) -> Dict:
+ def _add(self, func: Callable, rule: str, method: str, cors: bool, compress: bool, cache_control: Optional[str]):
+ self._routes.append(Route(method, self._build_rule_pattern(rule), func, cors, compress, cache_control))
+
+ @staticmethod
+ def _build_rule_pattern(rule: str):
+ rule_regex: str = re.sub(r"(<\w+>)", r"(?P\1.+)", rule)
+ return re.compile("^{}$".format(rule_regex))
+
+ def resolve(self, event, context) -> Dict[str, Any]:
  self.current_event = self._as_data_class(event)
  self.lambda_context = context
 
@@ -88,22 +99,14 @@ def resolve(self, event: Dict, context: LambdaContext) -> Dict:
 
  return {"statusCode": status_code, "headers": headers, "body": body, "isBase64Encoded": base64_encoded}
 
- def _append(self, func: Callable, rule: str, method: str, cors: bool, compress: bool, cache_control: Optional[str]):
- self._routes.append(RouteEntry(method, self._build_rule_pattern(rule), func, cors, compress, cache_control))
-
- @staticmethod
- def _build_rule_pattern(rule: str):
- rule_regex: str = re.sub(r"(<\w+>)", r"(?P\1.+)", rule)
- return re.compile("^{}$".format(rule_regex))
-
  def _as_data_class(self, event: Dict) -> BaseProxyEvent:
  if self._proxy_type == ProxyEventType.http_api_v1:
  return APIGatewayProxyEvent(event)
  if self._proxy_type == ProxyEventType.http_api_v2:
  return APIGatewayProxyEventV2(event)
  return ALBEvent(event)
 
- def _find_route(self, method: str, path: str) -> Tuple[RouteEntry, Dict]:
+ def _find_route(self, method: str, path: str) -> Tuple[Route, Dict]:
  method = method.upper()
  for route in self._routes:
  if method != route.method:

tests/functional/event_handler/test_api_gateway.py

@@ -50,21 +50,6 @@ def get_lambda():
  assert result["headers"]["Content-Type"] == "application/json"
 
 
-def test_include_rule_matching():
- app = ApiGatewayResolver()
-
- @app.get("/<name>/<my_id>")
- def get_lambda(my_id: str, name: str):
- assert name == "my"
- return 200, "plain/html", my_id
-
- result = app(load_event("apiGatewayProxyEvent.json"), {})
-
- assert result["statusCode"] == 200
- assert result["headers"]["Content-Type"] == "plain/html"
- assert result["body"] == "path"
-
-
 def test_api_gateway():
  app = ApiGatewayResolver(proxy_type=ProxyEventType.api_gateway)
 
@@ -96,24 +81,61 @@ def my_path():
  assert result["body"] == "tom"
 
 
+def test_include_rule_matching():
+ app = ApiGatewayResolver()
+
+ @app.get("/<name>/<my_id>")
+ def get_lambda(my_id: str, name: str):
+ assert name == "my"
+ return 200, "plain/html", my_id
+
+ result = app(load_event("apiGatewayProxyEvent.json"), {})
+
+ assert result["statusCode"] == 200
+ assert result["headers"]["Content-Type"] == "plain/html"
+ assert result["body"] == "path"
+
+
 def test_no_matches():
  app = ApiGatewayResolver()
 
  @app.get("/not_matching_get")
  def no_get_matching():
  raise RuntimeError()
 
- @app.put("/no_matching")
+ @app.post("/no_matching_post")
+ def no_post_matching():
+ raise RuntimeError()
+
+ @app.put("/no_matching_put")
  def no_put_matching():
  raise RuntimeError()
 
- @app.delete("/no_matching")
+ @app.delete("/no_matching_delete")
  def no_delete_matching():
  raise RuntimeError()
 
+ @app.patch("/no_matching_patch")
+ def no_patch_matching():
+ raise RuntimeError()
+
  def handler(event, context):
  app.resolve(event, context)
 
+ routes = app._routes
+ assert len(routes) == 5
+ for route in routes:
+ if route.func == no_get_matching:
+ assert route.method == "GET"
+ if route.func == no_post_matching:
+ assert route.method == "POST"
+ if route.func == no_put_matching:
+ assert route.method == "PUT"
+ if route.func == no_delete_matching:
+ assert route.method == "DELETE"
+ if route.func == no_patch_matching:
+ assert route.method == "PATCH"
+
  with pytest.raises(ValueError):
  handler(load_event("apiGatewayProxyEvent.json"), None)