feat(event-handler): add http ProxyEvent handler

aws_lambda_powertools/event_handler/api_gateway.py

@@ -0,0 +1,149 @@
+import base64
+import json
+import re
+import zlib
+from enum import Enum
+from typing import Any, Callable, Dict, List, Optional, Tuple, Union
+
+from aws_lambda_powertools.shared.json_encoder import Encoder
+from aws_lambda_powertools.utilities.data_classes import ALBEvent, APIGatewayProxyEvent, APIGatewayProxyEventV2
+from aws_lambda_powertools.utilities.data_classes.common import BaseProxyEvent
+from aws_lambda_powertools.utilities.typing import LambdaContext
+
+
+class ProxyEventType(Enum):
+    http_api_v1 = "APIGatewayProxyEvent"
+    http_api_v2 = "APIGatewayProxyEventV2"
+    alb_event = "ALBEvent"
+    api_gateway = http_api_v1
+
+
+class Route:
+    def __init__(
+        self, method: str, rule: Any, func: Callable, cors: bool, compress: bool, cache_control: Optional[str]
+    ):
+        self.method = method.upper()
+        self.rule = rule
+        self.func = func
+        self.cors = cors
+        self.compress = compress
+        self.cache_control = cache_control
+
+
+class Response:
+    def __init__(self, status_code: int, content_type: str, body: Union[str, bytes], headers: Dict = None):
+        self.status_code = status_code
+        self.body = body
+        self.base64_encoded = False
+        self.headers: Dict = headers or {}
+        self.headers.setdefault("Content-Type", content_type)
+
+    def add_cors(self, method: str):
+        self.headers["Access-Control-Allow-Origin"] = "*"
+        self.headers["Access-Control-Allow-Methods"] = method
+        self.headers["Access-Control-Allow-Credentials"] = "true"
+
+    def add_cache_control(self, cache_control: str):
+        self.headers["Cache-Control"] = cache_control if self.status_code == 200 else "no-cache"
+
+    def compress(self):
+        self.headers["Content-Encoding"] = "gzip"
+        if isinstance(self.body, str):
+            self.body = bytes(self.body, "utf-8")
+        gzip = zlib.compressobj(9, zlib.DEFLATED, zlib.MAX_WBITS | 16)
+        self.body = gzip.compress(self.body) + gzip.flush()
+
+    def to_dict(self) -> Dict[str, Any]:
+        if isinstance(self.body, bytes):
+            self.base64_encoded = True
+            self.body = base64.b64encode(self.body).decode()
+        return {
+            "statusCode": self.status_code,
+            "headers": self.headers,
+            "body": self.body,
+            "isBase64Encoded": self.base64_encoded,
+        }
+
+
+class ApiGatewayResolver:
+    current_event: BaseProxyEvent
+    lambda_context: LambdaContext
+
+    def __init__(self, proxy_type: Enum = ProxyEventType.http_api_v1):
+        self._proxy_type = proxy_type
+        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)
+
+    def post(self, rule: str, cors: bool = False, compress: bool = False, cache_control: str = None):
+        return self.route(rule, "POST", cors, compress, cache_control)
+
+    def put(self, rule: str, cors: bool = False, compress: bool = False, cache_control: str = None):
+        return self.route(rule, "PUT", cors, compress, cache_control)
+
+    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._routes.append(Route(method, self._compile_regex(rule), func, cors, compress, cache_control))
+            return func
+
+        return register_resolver
+
+    def resolve(self, event, context) -> Dict[str, Any]:
+        self.current_event = self._to_data_class(event)
+        self.lambda_context = context
+        route, args = self._find_route(self.current_event.http_method.upper(), self.current_event.path)
+        response = self.to_response(route.func(**args))
+
+        if route.cors:
+            response.add_cors(route.method)
+        if route.cache_control:
+            response.add_cache_control(route.cache_control)
+        if route.compress and "gzip" in (self.current_event.get_header_value("accept-encoding") or ""):
+            response.compress()
+
+        return response.to_dict()
+
+    @staticmethod
+    def to_response(result: Union[Tuple[int, str, Union[bytes, str]], Dict, Response]) -> Response:
+        if isinstance(result, Response):
+            return result
+        elif isinstance(result, dict):
+            return Response(
+                status_code=200,
+                content_type="application/json",
+                body=json.dumps(result, separators=(",", ":"), cls=Encoder),
+            )
+        else:  # Tuple[int, str, Union[bytes, str]]
+            return Response(*result)
+
+    @staticmethod
+    def _compile_regex(rule: str):
+        rule_regex: str = re.sub(r"(<\w+>)", r"(?P\1.+)", rule)
+        return re.compile("^{}$".format(rule_regex))
+
+    def _to_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[Route, Dict]:
+        for route in self._routes:
+            if method != route.method:
+                continue
+            match: Optional[re.Match] = route.rule.match(path)
+            if match:
+                return route, match.groupdict()
+
+        raise ValueError(f"No route found for '{method}.{path}'")
+
+    def __call__(self, event, context) -> Any:
+        return self.resolve(event, context)

aws_lambda_powertools/utilities/data_classes/alb_event.py

@@ -21,14 +21,6 @@ class ALBEvent(BaseProxyEvent):
     def request_context(self) -> ALBEventRequestContext:
         return ALBEventRequestContext(self._data)
 
-    @property
-    def http_method(self) -> str:
-        return self["httpMethod"]
-
-    @property
-    def path(self) -> str:
-        return self["path"]
-
     @property
     def multi_value_query_string_parameters(self) -> Optional[Dict[str, List[str]]]:
         return self.get("multiValueQueryStringParameters")

aws_lambda_powertools/utilities/data_classes/api_gateway_proxy_event.py

@@ -217,15 +217,6 @@ def version(self) -> str:
     def resource(self) -> str:
         return self["resource"]
 
-    @property
-    def path(self) -> str:
-        return self["path"]
-
-    @property
-    def http_method(self) -> str:
-        """The HTTP method used. Valid values include: DELETE, GET, HEAD, OPTIONS, PATCH, POST, and PUT."""
-        return self["httpMethod"]
-
     @property
     def multi_value_headers(self) -> Dict[str, List[str]]:
         return self["multiValueHeaders"]
@@ -446,3 +437,12 @@ def path_parameters(self) -> Optional[Dict[str, str]]:
     @property
     def stage_variables(self) -> Optional[Dict[str, str]]:
         return self.get("stageVariables")
+
+    @property
+    def path(self) -> str:
+        return self.raw_path
+
+    @property
+    def http_method(self) -> str:
+        """The HTTP method used. Valid values include: DELETE, GET, HEAD, OPTIONS, PATCH, POST, and PUT."""
+        return self.request_context.http.method

aws_lambda_powertools/utilities/data_classes/common.py

@@ -1,3 +1,4 @@
+import json
 from typing import Any, Dict, Optional
 
 
@@ -57,8 +58,23 @@ def is_base64_encoded(self) -> Optional[bool]:
 
     @property
     def body(self) -> Optional[str]:
+        """Submitted body of the request as a string"""
         return self.get("body")
 
+    @property
+    def json_body(self) -> Any:
+        """Parses the submitted body as json"""
+        return json.loads(self["body"])
+
+    @property
+    def path(self) -> str:
+        return self["path"]
+
+    @property
+    def http_method(self) -> str:
+        """The HTTP method used. Valid values include: DELETE, GET, HEAD, OPTIONS, PATCH, POST, and PUT."""
+        return self["httpMethod"]
+
     def get_query_string_value(self, name: str, default_value: Optional[str] = None) -> Optional[str]:
         """Get query string value by name