AWS API Gateway with Amazon Lambda integrations support

docs/integrations.rst

@@ -3,6 +3,59 @@ 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 Lambda functions handle events from API Gateway (Amazon API Gateway event format version 1.0 and 2.0).
+ * `AWS Lambda function URLs <https://docs.aws.amazon.com/lambda/latest/dg/lambda-urls.html>` where Lambda functions handle events from dedicated HTTP(S) endpoint (Amazon API Gateway event format version 2.0).
+
+Low level
+~~~~~~~~~
+
+You can use ``APIGatewayEventV2OpenAPIRequest`` as an API Gateway event (format version 2.0) request factory:
+
+.. code-block:: python
+
+   from openapi_core import unmarshal_request
+   from openapi_core.contrib.aws import APIGatewayEventV2OpenAPIRequest
+
+   openapi_request = APIGatewayEventV2OpenAPIRequest(event)
+   result = unmarshal_request(openapi_request, spec=spec)
+
+If you use format version 1.0, then import and use ``APIGatewayEventOpenAPIRequest`` as an API Gateway event (format version 1.0) request factory.
+
+You can use ``APIGatewayEventV2ResponseOpenAPIResponse`` as an API Gateway event (format version 2.0) response factory:
+
+.. code-block:: python
+
+   from openapi_core import unmarshal_response
+   from openapi_core.contrib.aws import APIGatewayEventV2ResponseOpenAPIResponse
+
+   openapi_response = APIGatewayEventV2ResponseOpenAPIResponse(response)
+   result = unmarshal_response(openapi_request, openapi_response, spec=spec)
+
+If you use format version 1.0, then import and use ``APIGatewayEventResponseOpenAPIResponse`` as an API Gateway event (format version 1.0) response factory.
+
+ANY method
+~~~~~~~~~~
+
+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,18 @@
+"""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.requests import APIGatewayEventV2OpenAPIRequest
+from openapi_core.contrib.aws.responses import (
+    APIGatewayEventResponseOpenAPIResponse,
+)
+from openapi_core.contrib.aws.responses import (
+    APIGatewayEventV2ResponseOpenAPIResponse,
+)
+
+__all__ = [
+    "APIGatewayEventOpenAPIRequest",
+    "APIGatewayEventV2OpenAPIRequest",
+    "APIGatewayEventResponseOpenAPIResponse",
+    "APIGatewayEventV2ResponseOpenAPIResponse",
+    "APIGatewayPathFinder",
+]

openapi_core/contrib/aws/datatypes.py

@@ -0,0 +1,74 @@
+from typing import Dict
+from typing import List
+from typing import Optional
+
+from pydantic import Field
+from pydantic.dataclasses import dataclass
+
+
+class APIGatewayEventConfig:
+    extra = "allow"
+
+
+@dataclass(config=APIGatewayEventConfig, frozen=True)
+class APIGatewayEvent:
+    """AWS API Gateway event"""
+
+    headers: Dict[str, str]
+
+    path: str
+    httpMethod: str
+    resource: str
+
+    queryStringParameters: Optional[Dict[str, str]] = None
+    isBase64Encoded: Optional[bool] = None
+    body: Optional[str] = None
+    pathParameters: Optional[Dict[str, str]] = None
+    stageVariables: Optional[Dict[str, str]] = None
+
+    multiValueHeaders: Optional[Dict[str, List[str]]] = None
+    version: Optional[str] = "1.0"
+    multiValueQueryStringParameters: Optional[Dict[str, List[str]]] = None
+
+
+@dataclass(config=APIGatewayEventConfig, frozen=True)
+class APIGatewayEventV2:
+    """AWS API Gateway event v2"""
+
+    headers: Dict[str, str]
+
+    version: str
+    routeKey: str
+    rawPath: str
+    rawQueryString: str
+
+    queryStringParameters: Optional[Dict[str, str]] = None
+    isBase64Encoded: Optional[bool] = None
+    body: Optional[str] = None
+    pathParameters: Optional[Dict[str, str]] = None
+    stageVariables: Optional[Dict[str, str]] = None
+
+    cookies: Optional[List[str]] = None
+
+
+@dataclass(config=APIGatewayEventConfig, frozen=True)
+class APIGatewayEventResponse:
+    """AWS API Gateway event response"""
+
+    body: str
+    isBase64Encoded: bool
+    statusCode: int
+    headers: Dict[str, str]
+    multiValueHeaders: Dict[str, List[str]]
+
+
+@dataclass(config=APIGatewayEventConfig, frozen=True)
+class APIGatewayEventV2Response:
+    """AWS API Gateway event v2 response"""
+
+    body: str
+    isBase64Encoded: bool = False
+    statusCode: int = 200
+    headers: Dict[str, str] = Field(
+        default_factory=lambda: {"content-type": "application/json"}
+    )

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,109 @@
+from typing import Dict
+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.datatypes import APIGatewayEventV2
+from openapi_core.contrib.aws.types import APIGatewayEventPayload
+from openapi_core.datatypes import RequestParameters
+
+
+class APIGatewayEventOpenAPIRequest:
+    """
+    Converts an API Gateway event payload to an OpenAPI request
+    """
+
+    def __init__(self, payload: APIGatewayEventPayload):
+        self.event = APIGatewayEvent(**payload)
+
+        self.parameters = RequestParameters(
+            query=ImmutableMultiDict(self.query_params),
+            header=Headers(self.event.headers),
+            cookie=ImmutableMultiDict(),
+        )
+
+    @property
+    def query_params(self) -> Dict[str, str]:
+        params = self.event.queryStringParameters
+        if params is None:
+            return {}
+        return params
+
+    @property
+    def proto(self) -> str:
+        return self.event.headers.get("X-Forwarded-Proto", "https")
+
+    @property
+    def host(self) -> str:
+        return self.event.headers["Host"]
+
+    @property
+    def host_url(self) -> str:
+        return "://".join([self.proto, self.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", "")
+
+
+class APIGatewayEventV2OpenAPIRequest:
+    """
+    Converts an API Gateway event v2 payload to an OpenAPI request
+    """
+
+    def __init__(self, payload: APIGatewayEventPayload):
+        self.event = APIGatewayEventV2(**payload)
+
+        self.parameters = RequestParameters(
+            query=ImmutableMultiDict(self.query_params),
+            header=Headers(self.event.headers),
+            cookie=ImmutableMultiDict(),
+        )
+
+    @property
+    def query_params(self) -> Dict[str, str]:
+        if self.event.queryStringParameters is None:
+            return {}
+        return self.event.queryStringParameters
+
+    @property
+    def proto(self) -> str:
+        return self.event.headers.get("x-forwarded-proto", "https")
+
+    @property
+    def host(self) -> str:
+        return self.event.headers["host"]
+
+    @property
+    def host_url(self) -> str:
+        return "://".join([self.proto, self.host])
+
+    @property
+    def path(self) -> str:
+        return self.event.rawPath
+
+    @property
+    def method(self) -> str:
+        return self.event.routeKey.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,83 @@
+from json import dumps
+from typing import Union
+
+from werkzeug.datastructures import Headers
+
+from openapi_core.contrib.aws.datatypes import APIGatewayEventResponse
+from openapi_core.contrib.aws.datatypes import APIGatewayEventV2Response
+from openapi_core.contrib.aws.types import APIGatewayEventResponsePayload
+
+APIGatewayEventV2ResponseType = Union[APIGatewayEventV2Response, dict, str]
+
+
+class APIGatewayEventResponseOpenAPIResponse:
+    """
+    Converts an API Gateway event response payload to an OpenAPI request
+    """
+
+    def __init__(self, payload: APIGatewayEventResponsePayload):
+        self.response = APIGatewayEventResponse(**payload)
+
+    @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 Headers(self.response.headers)
+
+    @property
+    def mimetype(self) -> str:
+        content_type = self.response.headers.get("Content-Type", "")
+        assert isinstance(content_type, str)
+        return content_type
+
+
+class APIGatewayEventV2ResponseOpenAPIResponse:
+    """
+    Converts an API Gateway event v2 response payload to an OpenAPI request
+    """
+
+    def __init__(self, payload: Union[APIGatewayEventResponsePayload, str]):
+        if not isinstance(payload, dict):
+            payload = self._construct_payload(payload)
+        elif "statusCode" not in payload:
+            body = dumps(payload)
+            payload = self._construct_payload(body)
+
+        self.response = APIGatewayEventV2Response(**payload)
+
+    @staticmethod
+    def _construct_payload(body: str) -> APIGatewayEventResponsePayload:
+        return {
+            "isBase64Encoded": False,
+            "statusCode": 200,
+            "headers": {
+                "content-type": "application/json",
+            },
+            "body": body,
+        }
+
+    @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 Headers(self.response.headers)
+
+    @property
+    def mimetype(self) -> str:
+        content_type = self.response.headers.get(
+            "content-type", "application/json"
+        )
+        assert isinstance(content_type, str)
+        return content_type

openapi_core/contrib/aws/types.py

@@ -0,0 +1,5 @@
+from typing import Any
+from typing import Dict
+
+APIGatewayEventPayload = Dict[str, Any]
+APIGatewayEventResponsePayload = Dict[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",
+]