Amazon Lambda using AWS API Gateway proxy support

Author: p1c2u

docs/integrations.rst

@@ -3,6 +3,48 @@ Integrations
 
 Openapi-core integrates with your popular libraries and frameworks. Each integration offers different levels of integration that help validate and unmarshal your request and response data.
 
+AWS Lambda function using Amazon API Gateway Proxy
+--------------------------------------------------
+
+This section describes integration with `AWS Lambda <https://docs.aws.amazon.com/lambda/>`__ function using `Amazon API Gateway Proxy <hhttps://docs.aws.amazon.com/lambda/latest/dg/services-apigateway.html#apigateway-proxy>`__.
+
+Low level
+~~~~~~~~~
+
+You can use ``APIGatewayProxyEventOpenAPIRequest`` as an API Gateway proxy event (REST API) request factory:
+
+.. code-block:: python
+
+   from openapi_core import unmarshal_request
+   from openapi_core.contrib.aws import APIGatewayProxyEventOpenAPIRequest
+
+   openapi_request = APIGatewayProxyEventOpenAPIRequest(lambda_event)
+   result = unmarshal_request(openapi_request, spec=spec)
+
+You can use ``APIGatewayProxyEventOpenAPIResponse`` as an API Gateway proxy event (REST API) response factory:
+
+.. code-block:: python
+
+   from openapi_core import unmarshal_response
+   from openapi_core.contrib.aws import APIGatewayProxyEventOpenAPIResponse
+
+   openapi_response = APIGatewayProxyEventOpenAPIResponse(lambda_response)
+   result = unmarshal_response(openapi_request, openapi_response, spec=spec)
+
+If you use ``x-amazon-apigateway-any-method`` OpenAPI extension with ANY method that catches all HTTP methods, you want to define ``path_finder_cls`` to be ``AmazonAPIGatewayPathFinder``:
+
+.. code-block:: python
+
+   from openapi_core.contrib.aws import AmazonAPIGatewayPathFinder
+
+   result = unmarshal_response(
+      openapi_request,
+      openapi_response,
+      spec=spec,
+      path_finder_cls=AmazonAPIGatewayPathFinder,
+   )
+
+
 Bottle
 ------
 

openapi_core/contrib/aws/__init__.py

@@ -0,0 +1,14 @@
+"""OpenAPI core contrib aws module"""
+from openapi_core.contrib.aws.finders import AmazonAPIGatewayPathFinder
+from openapi_core.contrib.aws.requests import (
+    APIGatewayProxyEventOpenAPIRequest,
+)
+from openapi_core.contrib.aws.responses import (
+    APIGatewayProxyEventOpenAPIResponse,
+)
+
+__all__ = [
+    "AmazonAPIGatewayPathFinder",
+    "APIGatewayProxyEventOpenAPIRequest",
+    "APIGatewayProxyEventOpenAPIResponse",
+]

openapi_core/contrib/aws/datatypes.py

@@ -0,0 +1,60 @@
+from typing import List
+from typing import Optional
+
+from pydantic.dataclasses import dataclass
+
+
+class LambdaConfig:
+    extra = "allow"
+
+
+@dataclass(config=LambdaConfig, kw_only=True)
+class BaseAPIGatewayProxyEvent:
+    path: str
+    httpMethod: str
+    headers: dict
+    queryStringParameters: Optional[dict] = None
+    isBase64Encoded: Optional[bool] = None
+    body: Optional[str] = None
+    pathParameters: Optional[dict] = None
+    stageVariables: Optional[dict] = None
+
+
+@dataclass(config=LambdaConfig, kw_only=True)
+class APIGatewayProxyEvent(BaseAPIGatewayProxyEvent):
+    """AWS API Gateway proxy v1 event"""
+
+    resource: str
+    multiValueHeaders: dict
+    version: Optional[str] = "1.0"
+    multiValueQueryStringParameters: Optional[dict] = None
+
+
+@dataclass(config=LambdaConfig, kw_only=True)
+class APIGatewayProxyV2Event(BaseAPIGatewayProxyEvent):
+    """AWS API Gateway proxy v2 event"""
+
+    version: str
+    routeKey: str
+    rawPath: dict
+    rawQueryString: str
+    cookies: Optional[List[str]] = None
+
+
+@dataclass(config=LambdaConfig, kw_only=True)
+class APIGatewayProxyEventResponse:
+    """AWS API Gateway proxy v1 event response"""
+
+    isBase64Encoded: bool
+    statusCode: str
+    headers: dict
+    multiValueHeaders: dict
+    body: str
+
+
+@dataclass(config=LambdaConfig, kw_only=True)
+class APIGatewayProxyV2EventResponse:
+    """AWS API Gateway proxy v2 event response"""
+
+    isBase64Encoded: bool = False
+    statusCode: str = 200

openapi_core/contrib/aws/finders.py

@@ -0,0 +1,8 @@
+from openapi_core.templating.paths.finders import APICallPathFinder
+from openapi_core.templating.paths.iterators import AnyMethodOperationsIterator
+
+
+class AmazonAPIGatewayPathFinder(APICallPathFinder):
+    operations_iterator = AnyMethodOperationsIterator(
+        any_method="x-amazon-apigateway-any-method",
+    )

openapi_core/contrib/aws/requests.py

@@ -0,0 +1,41 @@
+from typing import Optional
+
+from werkzeug.datastructures import Headers
+from werkzeug.datastructures import ImmutableMultiDict
+
+from openapi_core.contrib.aws.datatypes import APIGatewayProxyEvent
+from openapi_core.contrib.aws.typing import LambdaEvent
+from openapi_core.datatypes import RequestParameters
+
+
+class APIGatewayProxyEventOpenAPIRequest:
+    def __init__(self, lambda_event: LambdaEvent):
+        self.event = APIGatewayProxyEvent(**lambda_event)
+
+        self.parameters = RequestParameters(
+            query=ImmutableMultiDict(self.event.queryStringParameters),
+            header=Headers(self.event.headers),
+            cookie=ImmutableMultiDict(),
+        )
+
+    @property
+    def host_url(self) -> str:
+        proto = self.event.headers["X-Forwarded-Proto"]
+        host = self.event.headers["Host"]
+        return "://".join([proto, host])
+
+    @property
+    def path(self) -> str:
+        return self.event.resource
+
+    @property
+    def method(self) -> str:
+        return self.event.httpMethod.lower()
+
+    @property
+    def body(self) -> Optional[str]:
+        return self.event.body
+
+    @property
+    def mimetype(self) -> str:
+        return self.event.headers.get("Content-Type", "")

openapi_core/contrib/aws/responses.py

@@ -0,0 +1,32 @@
+from werkzeug.datastructures import Headers
+
+from openapi_core.contrib.aws.datatypes import APIGatewayProxyEventResponse
+from openapi_core.contrib.aws.typing import LambdaResponse
+
+
+class APIGatewayProxyEventOpenAPIResponse:
+    """Lambda response
+
+    Documentation: https://docs.aws.amazon.com/apigateway/latest/developerguide/http-api-develop-integrations-lambda.html
+    """
+
+    def __init__(self, lambda_response: LambdaResponse):
+        self.response = APIGatewayProxyEventResponse(**lambda_response)
+
+    @property
+    def data(self) -> str:
+        return self.response.body
+
+    @property
+    def status_code(self) -> int:
+        return self.response.statusCode
+
+    @property
+    def headers(self) -> Headers:
+        return self.response.headers
+
+    @property
+    def mimetype(self) -> str:
+        content_type = self.response.headers.get("Content-Type", "")
+        assert isinstance(content_type, str)
+        return content_type

openapi_core/contrib/aws/typing.py

@@ -0,0 +1,5 @@
+from typing import Any
+from typing import Mapping
+
+LambdaEvent = Mapping[str, Any]
+LambdaResponse = Mapping[str, Any]

openapi_core/templating/paths/__init__.py

@@ -0,0 +1,7 @@
+from openapi_core.templating.paths.finders import APICallPathFinder
+from openapi_core.templating.paths.finders import WebhookPathFinder
+
+__all__ = [
+    "APICallPathFinder",
+    "WebhookPathFinder",
+]