Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

AWS API Gateway with Amazon Lambda integrations support#545

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to ourterms of service andprivacy statement. We’ll occasionally send you account related emails.

Already on GitHub?Sign in to your account

Closed
Show file tree
Hide file tree
Changes fromall commits
Commits
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 53 additions & 0 deletionsdocs/integrations.rst
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -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
------

Expand Down
22 changes: 22 additions & 0 deletionsopenapi_core/contrib/aws/__init__.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
@@ -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",
]
74 changes: 74 additions & 0 deletionsopenapi_core/contrib/aws/datatypes.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
@@ -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"}
)
18 changes: 18 additions & 0 deletionsopenapi_core/contrib/aws/finders.py
View file
Open in desktop
Original file line numberDiff line numberDiff 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",
)
151 changes: 151 additions & 0 deletionsopenapi_core/contrib/aws/requests.py
View file
Open in desktop
Original file line numberDiff line numberDiff 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()
Loading
[8]ページ先頭
©2009-2025 Movatter.jp