AWS API Gateway with Amazon Lambda integrations support

docs/integrations/aws.rst

@@ -0,0 +1,70 @@
+Amazon API Gateway
+==================
+
+This section describes integration with `Amazon API Gateway <https://aws.amazon.com/api-gateway/>`__.
+
+It is useful for:
+
+* `AWS Proxy integrations (i.e. AWS Lambda) for HTTP API <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).
+* `HTTP Proxy integrations for HTTP API <https://docs.aws.amazon.com/apigateway/latest/developerguide/http-api-develop-integrations-http.html>` where HTTP service handle events from API Gateway.
+* `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.
+
+AWS Proxy event
+^^^^^^^^^^^^^^^
+
+Use ``APIGatewayAWSProxyV2OpenAPIRequest`` to create OpenAPI request from an API Gateway event (format version 2.0):
+
+.. code-block:: python
+
+    from openapi_core.contrib.aws import APIGatewayAWSProxyV2OpenAPIRequest
+
+    def handler(event, context):
+        openapi_request = APIGatewayAWSProxyV2OpenAPIRequest(event)
+        result = openapi.unmarshal_request(openapi_request)
+        return {
+            "statusCode": 200,
+            "body": "Hello world",
+        }
+
+If you use format version 1.0, then import and use ``APIGatewayAWSProxyOpenAPIRequest``.
+
+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/__init__.py

@@ -0,0 +1,17 @@
+"""OpenAPI core contrib django module"""
+
+from openapi_core.contrib.aws.finders import APIGatewayPathFinder
+from openapi_core.contrib.aws.requests import APIGatewayAWSProxyOpenAPIRequest
+from openapi_core.contrib.aws.requests import APIGatewayAWSProxyV2OpenAPIRequest
+from openapi_core.contrib.aws.requests import APIGatewayHTTPProxyOpenAPIRequest
+from openapi_core.contrib.aws.responses import APIGatewayEventResponseOpenAPIResponse
+from openapi_core.contrib.aws.responses import APIGatewayEventV2ResponseOpenAPIResponse
+
+__all__ = [
+    "APIGatewayPathFinder",
+    "APIGatewayAWSProxyOpenAPIRequest",
+    "APIGatewayAWSProxyV2OpenAPIRequest",
+    "APIGatewayHTTPProxyOpenAPIRequest",
+    "APIGatewayEventResponseOpenAPIResponse",
+    "APIGatewayEventV2ResponseOpenAPIResponse",
+]

openapi_core/contrib/aws/datatypes.py

@@ -0,0 +1,104 @@
+from typing import Dict
+from typing import List
+from typing import Literal
+from typing import Optional
+
+from pydantic import Field
+from pydantic.dataclasses import dataclass
+
+API_GATEWAY_EVENT_CONFIG = dict(extra="allow")
+
+
+@dataclass(frozen=True)
+class APIGatewayEventRequestContext:
+    """AWS API Gateway event request context"""
+    model_config = API_GATEWAY_EVENT_CONFIG
+
+    resourceId: str
+
+
+@dataclass(frozen=True)
+class APIGatewayEvent:
+    """AWS API Gateway event"""
+    model_config = API_GATEWAY_EVENT_CONFIG
+
+    headers: Dict[str, str]
+
+    path: str
+    httpMethod: str
+    resource: str
+    requestContext: APIGatewayEventRequestContext
+
+    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(frozen=True)
+class APIGatewayEventV2Http:
+    """AWS API Gateway event v2 HTTP"""
+    model_config = API_GATEWAY_EVENT_CONFIG
+
+    method: str
+    path: str
+
+
+@dataclass(frozen=True)
+class APIGatewayEventV2RequestContext:
+    """AWS API Gateway event v2 request context"""
+    model_config = API_GATEWAY_EVENT_CONFIG
+
+    http: APIGatewayEventV2Http
+
+
+@dataclass(frozen=True)
+class APIGatewayEventV2:
+    """AWS API Gateway event v2"""
+    model_config = API_GATEWAY_EVENT_CONFIG
+
+    headers: Dict[str, str]
+
+    version: Literal["2.0"]
+    routeKey: str
+    rawPath: str
+    rawQueryString: str
+    requestContext: APIGatewayEventV2RequestContext
+
+    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(frozen=True)
+class APIGatewayEventResponse:
+    """AWS API Gateway event response"""
+    model_config = API_GATEWAY_EVENT_CONFIG
+
+    body: str
+    isBase64Encoded: bool
+    statusCode: int
+    headers: Dict[str, str]
+    multiValueHeaders: Dict[str, List[str]]
+
+
+@dataclass(frozen=True)
+class APIGatewayEventV2Response:
+    """AWS API Gateway event v2 response"""
+    model_config = API_GATEWAY_EVENT_CONFIG
+
+    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,11 @@
+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",
+    )

openapi_core/contrib/aws/requests.py

@@ -0,0 +1,174 @@
+from typing import Dict
+from typing import Optional
+from typing import Tuple
+from urllib.parse import urljoin
+
+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.contrib.aws.util import parse_forwarded
+from openapi_core.datatypes import RequestParameters
+
+
+class APIGatewayAWSProxyOpenAPIRequest:
+    """
+    API Gateway AWS proxy event payload 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
+    requestContext's resourceId. 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 resource_id(self) -> Tuple[str, str]:
+        return self.event.requestContext.resourceId.split(" ")
+
+    @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.resource_id[0].lower()
+
+    @property
+    def body(self) -> Optional[str]:
+        return self.event.body
+
+    @property
+    def content_type(self) -> str:
+        return self.event.headers.get("Content-Type", "")
+
+    @property
+    def path_pattern(self):
+        return self.resource_id[1]
+
+
+class APIGatewayAWSProxyV2OpenAPIRequest:
+    """
+    API Gateway AWS Proxy event v2 payload 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:
+        # use Forwarded header if available
+        if "forwarded" in self.event.headers:
+            forwarded = parse_forwarded(self.event.headers["forwarded"])
+            if "host" in forwarded:
+                return forwarded["host"]
+        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 content_type(self) -> str:
+        return self.event.headers.get("content-type", "")
+
+
+class APIGatewayHTTPProxyOpenAPIRequest(APIGatewayAWSProxyV2OpenAPIRequest):
+    """
+    API Gateway HTTP proxy integration event payload OpenAPI request.
+
+    Uses http integration path and method data.
+
+    NOTE: If you use HTTP integration not in root path then you need to provide the base path
+    otherwise it won't find the correct path because it is not send with event.
+    """
+
+    def __init__(self, payload: APIGatewayEventPayload, base_path: str = "/"):
+        super().__init__(payload)
+        self.base_path = base_path
+
+    @property
+    def path(self) -> str:
+        return urljoin(self.base_path, self.event.requestContext.http.path.lstrip('/'))
+
+    @property
+    def method(self) -> str:
+        return self.event.requestContext.http.method.lower()