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,22 @@
+"""OpenAPI core contrib aws module"""
+from openapi_core.contrib.aws.decorators import (
+    APIGatewayEventV2OpenAPIHandleDecorator,
+)
+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",
+    "APIGatewayEventResponseOpenAPIResponse",
+    "APIGatewayEventV2OpenAPIHandleDecorator",
+    "APIGatewayEventV2OpenAPIRequest",
+    "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,18 @@
+from openapi_core.templating.paths.finders import APICallPathFinder
+from openapi_core.templating.paths.iterators import (
+    CatchAllMethodOperationsIterator,
+)
+
+
+class APIGatewayPathFinder(APICallPathFinder):
+    operations_iterator = CatchAllMethodOperationsIterator(
+        "any",
+        "x-amazon-apigateway-any-method",
+    )
+
+
+class APIGatewayIntegrationPathFinder(APICallPathFinder):
+    operations_iterator = CatchAllMethodOperationsIterator(
+        "any",
+        "x-amazon-apigateway-any-method",
+    )

openapi_core/contrib/aws/requests.py

@@ -0,0 +1,151 @@
+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.
+
+    Designed to be used with API Gateway REST API specification exports for
+    integrations that use event v1 payload. Uses API Gateway event v1 httpMethod
+    and path data. Requires APIGatewayPathFinder to resolve ANY methods.
+    """
+
+    def __init__(self, payload: APIGatewayEventPayload):
+        self.event = APIGatewayEvent(**payload)
+
+        self.parameters = RequestParameters(
+            path=self.path_params,
+            query=ImmutableMultiDict(self.query_params),
+            header=Headers(self.event.headers),
+            cookie=ImmutableMultiDict(),
+        )
+
+    @property
+    def path_params(self) -> Dict[str, str]:
+        params = self.event.pathParameters
+        if params is None:
+            return {}
+        return params
+
+    @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.path
+
+    @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.
+
+    Designed to be used with API Gateway HTTP API specification exports for
+    integrations that use event v2 payload. Uses API Gateway event v2 routeKey
+    and rawPath data. Requires APIGatewayPathFinder to resolve ANY methods.
+
+    .. note::
+       API Gateway HTTP APIs don't support request validation
+    """
+
+    def __init__(self, payload: APIGatewayEventPayload):
+        self.event = APIGatewayEventV2(**payload)
+
+        self.parameters = RequestParameters(
+            path=self.path_params,
+            query=ImmutableMultiDict(self.query_params),
+            header=Headers(self.event.headers),
+            cookie=ImmutableMultiDict(),
+        )
+
+    @property
+    def path_params(self) -> Dict[str, str]:
+        if self.event.pathParameters is None:
+            return {}
+        return self.event.pathParameters
+
+    @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.split(" ")[0].lower()
+
+    @property
+    def body(self) -> Optional[str]:
+        return self.event.body
+
+    @property
+    def mimetype(self) -> str:
+        return self.event.headers.get("content-type", "")
+
+
+class APIGatewayEventV2HTTPOpenAPIRequest(APIGatewayEventV2OpenAPIRequest):
+    """
+    Converts an API Gateway event v2 payload to an OpenAPI request.
+
+    Uses http integration path and method data.
+    """
+
+    @property
+    def path(self) -> str:
+        return self.event.http.path
+
+    @property
+    def method(self) -> str:
+        return self.event.http.method.lower()