AWS API Gateway with Amazon Lambda integrations support

docs/integrations/aws.rst

@@ -0,0 +1,69 @@
+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).
+
+ANY method
+----------
+
+Amazon API Gateway defines special ``ANY`` method that catches all HTTP methods. It is specified as `x-amazon-apigateway-any-method <https://docs.aws.amazon.com/apigateway/latest/developerguide/api-gateway-swagger-extensions-any-method.html>`__ OpenAPI extension. The extension is handled within custom path finder and can be used by setting ``path_finder_cls`` to be ``APIGatewayPathFinder``:
+
+.. code-block:: python
+  :emphasize-lines: 1,4
+
+    from openapi_core.contrib.aws import APIGatewayPathFinder
+
+    config = Config(
+        path_finder_cls=APIGatewayPathFinder,
+    )
+    openapi = OpenAPI.from_file_path('openapi.json', config=config)
+
+Low level
+---------
+
+The integration defines classes useful for low level integration.
+
+Request
+^^^^^^^
+
+Use ``APIGatewayEventV2OpenAPIRequest`` to create OpenAPI request from an API Gateway event (format version 2.0):
+
+.. code-block:: python
+
+    from openapi_core.contrib.aws import APIGatewayEventV2OpenAPIRequest
+
+    def handler(event, context):
+        openapi_request = APIGatewayEventV2OpenAPIRequest(event)
+        result = openapi.unmarshal_request(openapi_request)
+        return {
+            "statusCode": 200,
+            "body": "Hello world",
+        }
+
+If you use format version 1.0, then import and use ``APIGatewayEventOpenAPIRequest``.
+
+Response
+^^^^^^^^
+
+Use ``APIGatewayEventV2ResponseOpenAPIResponse`` to create OpenAPI response from API Gateway event (format version 2.0) response:
+
+.. code-block:: python
+
+    from openapi_core.contrib.aws import APIGatewayEventV2ResponseOpenAPIResponse
+
+    def handler(event, context):
+        openapi_request = APIGatewayEventV2OpenAPIRequest(event)
+        response = {
+            "statusCode": 200,
+            "body": "Hello world",
+        }
+        openapi_response = APIGatewayEventV2ResponseOpenAPIResponse(response)
+        result = openapi.unmarshal_response(openapi_request, openapi_response)
+        return response
+
+If you use format version 1.0, then import and use ``APIGatewayEventResponseOpenAPIResponse``.

docs/integrations/index.rst

@@ -6,6 +6,7 @@ Openapi-core integrates with your popular libraries and frameworks. Each integra
 .. toctree::
     :maxdepth: 1
 
+    aws
     aiohttp
     bottle
     django

openapi_core/contrib/aws/datatypes.py

@@ -0,0 +1,83 @@
+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 APIGatewayEventV2Http:
+    """AWS API Gateway event v2 HTTP"""
+
+    method: str
+    path: str
+
+
+@dataclass(config=APIGatewayEventConfig, frozen=True)
+class APIGatewayEventV2:
+    """AWS API Gateway event v2"""
+
+    headers: Dict[str, str]
+
+    version: str
+    routeKey: str
+    rawPath: str
+    rawQueryString: str
+    http: APIGatewayEventV2Http
+
+    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()