AWS API Gateway with Amazon Lambda integrations 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.
 
+Amazon API Gateway
+------------------
+
+This section describes integration with `Amazon API Gateway <https://aws.amazon.com/api-gateway/>`__. It is useful for `AWS Lambda integrations <https://docs.aws.amazon.com/apigateway/latest/developerguide/http-api-develop-integrations-lambda.html>`__ where functions handle events from API Gateway.
+
+Low level
+~~~~~~~~~
+
+You can use ``APIGatewayEventOpenAPIRequest`` as an API Gateway event request factory:
+
+.. code-block:: python
+
+   from openapi_core import unmarshal_request
+   from openapi_core.contrib.aws import APIGatewayEventOpenAPIRequest
+
+   openapi_request = APIGatewayEventOpenAPIRequest(event)
+   result = unmarshal_request(openapi_request, spec=spec)
+
+You can use ``APIGatewayEventResponseOpenAPIResponse`` as an API Gateway (HTTP API) event response factory:
+
+.. code-block:: python
+
+   from openapi_core import unmarshal_response
+   from openapi_core.contrib.aws import APIGatewayEventResponseOpenAPIResponse
+
+   openapi_response = APIGatewayEventResponseOpenAPIResponse(response)
+   result = unmarshal_response(openapi_request, openapi_response, spec=spec)
+
+API Gateway have special ``ANY`` method that catches all HTTP methods. It's specified as `x-amazon-apigateway-any-method <https://docs.aws.amazon.com/apigateway/latest/developerguide/api-gateway-swagger-extensions-any-method.html>__` OpenAPI extension. If you use the extension, you want to define ``path_finder_cls`` to be ``APIGatewayPathFinder``:
+
+.. code-block:: python
+
+   from openapi_core.contrib.aws import APIGatewayPathFinder
+
+   result = unmarshal_response(
+      openapi_request,
+      openapi_response,
+      spec=spec,
+      path_finder_cls=APIGatewayPathFinder,
+   )
+
+
 Bottle
 ------
 

openapi_core/contrib/aws/__init__.py

@@ -0,0 +1,12 @@
+"""OpenAPI core contrib aws module"""
+from openapi_core.contrib.aws.finders import APIGatewayPathFinder
+from openapi_core.contrib.aws.requests import APIGatewayEventOpenAPIRequest
+from openapi_core.contrib.aws.responses import (
+    APIGatewayEventResponseOpenAPIResponse,
+)
+
+__all__ = [
+    "APIGatewayEventOpenAPIRequest",
+    "APIGatewayEventResponseOpenAPIResponse",
+    "APIGatewayPathFinder",
+]

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 BaseAPIGatewayEvent:
+    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 APIGatewayEvent(BaseAPIGatewayEvent):
+    """AWS API Gateway event"""
+
+    resource: str
+    multiValueHeaders: dict
+    version: Optional[str] = "1.0"
+    multiValueQueryStringParameters: Optional[dict] = None
+
+
+@dataclass(config=LambdaConfig, kw_only=True)
+class APIGatewayEventV2(BaseAPIGatewayEvent):
+    """AWS API Gateway event v2"""
+
+    version: str
+    routeKey: str
+    rawPath: dict
+    rawQueryString: str
+    cookies: Optional[List[str]] = None
+
+
+@dataclass(config=LambdaConfig, kw_only=True)
+class APIGatewayEventResponse:
+    """AWS API Gateway event response"""
+
+    isBase64Encoded: bool
+    statusCode: str
+    headers: dict
+    multiValueHeaders: dict
+    body: str
+
+
+@dataclass(config=LambdaConfig, kw_only=True)
+class APIGatewayEventV2Response:
+    """AWS API Gateway event v2 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 APIGatewayPathFinder(APICallPathFinder):
+    operations_iterator = AnyMethodOperationsIterator(
+        any_method="x-amazon-apigateway-any-method",
+    )

openapi_core/contrib/aws/requests.py

@@ -0,0 +1,45 @@
+from typing import Optional
+
+from werkzeug.datastructures import Headers
+from werkzeug.datastructures import ImmutableMultiDict
+
+from openapi_core.contrib.aws.datatypes import APIGatewayEvent
+from openapi_core.contrib.aws.typing import APIGatewayEventDict
+from openapi_core.datatypes import RequestParameters
+
+
+class APIGatewayEventOpenAPIRequest:
+    """
+    Converts an API Gateway event to an OpenAPI request
+    """
+
+    def __init__(self, event: APIGatewayEventDict):
+        self.event = APIGatewayEvent(**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,31 @@
+from werkzeug.datastructures import Headers
+
+from openapi_core.contrib.aws.datatypes import APIGatewayEventResponse
+from openapi_core.contrib.aws.typing import APIGatewayEventResponseDict
+
+
+class APIGatewayEventResponseOpenAPIResponse:
+    """
+    Converts an API Gateway event response to an OpenAPI request
+    """
+
+    def __init__(self, response: APIGatewayEventResponseDict):
+        self.response = APIGatewayEventResponse(**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
+
+APIGatewayEventDict = Mapping[str, Any]
+APIGatewayEventResponseDict = 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",
+]