-
-
Notifications
You must be signed in to change notification settings - Fork 132
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
AWS API Gateway with Amazon Lambda integrations support
- Loading branch information
Showing
10 changed files
with
501 additions
and
114 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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``. |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Empty file.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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"} | ||
) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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", | ||
) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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() |
Oops, something went wrong.