github/openapi-core
1
0
Fork 0
mirror of https://github.com/correl/openapi-core.git synced 2024-12-28 19:19:23 +00:00
Code Issues Projects Releases Packages Wiki Activity

Merge pull request #215 from p1c2u/feature/falcon-inteagration

Browse source 
Falcon integration
This commit is contained in:
A 2020-03-11 11:23:43 +00:00 committed by GitHub
commit eef515f729
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
13 changed files with 576 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

42
README.rst
View file

@ -206,6 +206,48 @@ You can use DjangoOpenAPIResponse as a Django response factory:
validator = ResponseValidator(spec)
result = validator.validate(openapi_request, openapi_response)
Falcon
******
This section describes integration with `Falcon <https://falconframework.org>`__ web framework.
Middleware
==========
Falcon API can be integrated by `FalconOpenAPIMiddleware` middleware.
.. code-block:: python
from openapi_core.contrib.falcon.middlewares import FalconOpenAPIMiddleware
openapi_middleware = FalconOpenAPIMiddleware.from_spec(spec)
api = falcon.API(middleware=[openapi_middleware])
Low level
=========
For Falcon you can use FalconOpenAPIRequest a Falcon request factory:
.. code-block:: python
from openapi_core.validation.request.validators import RequestValidator
from openapi_core.contrib.falcon import FalconOpenAPIRequest
openapi_request = FalconOpenAPIRequest(falcon_request)
validator = RequestValidator(spec)
result = validator.validate(openapi_request)
You can use FalconOpenAPIResponse as a Falcon response factory:
.. code-block:: python
from openapi_core.validation.response.validators import ResponseValidator
from openapi_core.contrib.falcon import FalconOpenAPIResponse
openapi_response = FalconOpenAPIResponse(falcon_response)
validator = ResponseValidator(spec)
result = validator.validate(openapi_request, openapi_response)
Flask
*****

5
openapi_core/contrib/falcon/__init__.py Normal file
View file

@ -0,0 +1,5 @@
from openapi_core.contrib.falcon.requests import FalconOpenAPIRequestFactory
from openapi_core.contrib.falcon.responses import FalconOpenAPIResponseFactory
__all__ = ["FalconOpenAPIRequestFactory", "FalconOpenAPIResponseFactory"]

52
openapi_core/contrib/falcon/handlers.py Normal file
View file

@ -0,0 +1,52 @@
"""OpenAPI core contrib falcon handlers module"""
from json import dumps
from falcon.constants import MEDIA_JSON
from falcon.status_codes import (
HTTP_400, HTTP_404, HTTP_405, HTTP_415,
)
from openapi_core.schema.media_types.exceptions import InvalidContentType
from openapi_core.templating.paths.exceptions import (
ServerNotFound, OperationNotFound, PathNotFound,
)
class FalconOpenAPIErrorsHandler(object):
OPENAPI_ERROR_STATUS = {
ServerNotFound: 400,
OperationNotFound: 405,
PathNotFound: 404,
InvalidContentType: 415,
}
FALCON_STATUS_CODES = {
400: HTTP_400,
404: HTTP_404,
405: HTTP_405,
415: HTTP_415,
}
@classmethod
def handle(cls, req, resp, errors):
data_errors = [
cls.format_openapi_error(err)
for err in errors
]
data = {
'errors': data_errors,
}
data_error_max = max(data_errors, key=lambda x: x['status'])
resp.content_type = MEDIA_JSON
resp.status = cls.FALCON_STATUS_CODES.get(
data_error_max['status'], HTTP_400)
resp.body = dumps(data)
resp.complete = True
@classmethod
def format_openapi_error(cls, error):
return {
'title': str(error),
'status': cls.OPENAPI_ERROR_STATUS.get(error.__class__, 400),
'class': str(type(error)),
}

73
openapi_core/contrib/falcon/middlewares.py Normal file
View file

@ -0,0 +1,73 @@
"""OpenAPI core contrib falcon middlewares module"""
from openapi_core.contrib.falcon.handlers import FalconOpenAPIErrorsHandler
from openapi_core.contrib.falcon.requests import FalconOpenAPIRequestFactory
from openapi_core.contrib.falcon.responses import FalconOpenAPIResponseFactory
from openapi_core.validation.processors import OpenAPIProcessor
from openapi_core.validation.request.validators import RequestValidator
from openapi_core.validation.response.validators import ResponseValidator
class FalconOpenAPIMiddleware(OpenAPIProcessor):
def __init__(
self,
request_validator,
response_validator,
request_factory,
response_factory,
openapi_errors_handler,
):
super(FalconOpenAPIMiddleware, self).__init__(
request_validator, response_validator)
self.request_factory = request_factory
self.response_factory = response_factory
self.openapi_errors_handler = openapi_errors_handler
def process_request(self, req, resp):
openapi_req = self._get_openapi_request(req)
req_result = super(FalconOpenAPIMiddleware, self).process_request(
openapi_req)
if req_result.errors:
return self._handle_request_errors(req, resp, req_result)
req.openapi = req_result
def process_response(self, req, resp, resource, req_succeeded):
openapi_req = self._get_openapi_request(req)
openapi_resp = self._get_openapi_response(resp)
resp_result = super(FalconOpenAPIMiddleware, self).process_response(
openapi_req, openapi_resp)
if resp_result.errors:
return self._handle_response_errors(req, resp, resp_result)
def _handle_request_errors(self, req, resp, request_result):
return self.openapi_errors_handler.handle(
req, resp, request_result.errors)
def _handle_response_errors(self, req, resp, response_result):
return self.openapi_errors_handler.handle(
req, resp, response_result.errors)
def _get_openapi_request(self, request):
return self.request_factory.create(request)
def _get_openapi_response(self, response):
return self.response_factory.create(response)
@classmethod
def from_spec(
cls,
spec,
request_factory=FalconOpenAPIRequestFactory,
response_factory=FalconOpenAPIResponseFactory,
openapi_errors_handler=FalconOpenAPIErrorsHandler,
):
request_validator = RequestValidator(spec)
response_validator = ResponseValidator(spec)
return cls(
request_validator=request_validator,
response_validator=response_validator,
request_factory=request_factory,
response_factory=response_factory,
openapi_errors_handler=openapi_errors_handler,
)

45
openapi_core/contrib/falcon/requests.py Normal file
View file

@ -0,0 +1,45 @@
"""OpenAPI core contrib falcon responses module"""
from json import dumps
from werkzeug.datastructures import ImmutableMultiDict
from openapi_core.validation.request.datatypes import (
OpenAPIRequest, RequestParameters,
)
class FalconOpenAPIRequestFactory:
@classmethod
def create(cls, request):
"""
Create OpenAPIRequest from falcon Request and route params.
"""
method = request.method.lower()
# gets deduced by path finder against spec
path = {}
# Support falcon-jsonify.
body = (
dumps(request.json) if getattr(request, "json", None)
else request.bounded_stream.read()
)
mimetype = request.options.default_media_type
if request.content_type:
mimetype = request.content_type.partition(";")[0]
query = ImmutableMultiDict(request.params.items())
parameters = RequestParameters(
query=query,
header=request.headers,
cookie=request.cookies,
path=path,
)
return OpenAPIRequest(
full_url_pattern=request.url,
method=method,
parameters=parameters,
body=body,
mimetype=mimetype,
)

20
openapi_core/contrib/falcon/responses.py Normal file
View file

@ -0,0 +1,20 @@
"""OpenAPI core contrib falcon responses module"""
from openapi_core.validation.response.datatypes import OpenAPIResponse
class FalconOpenAPIResponseFactory(object):
@classmethod
def create(cls, response):
status_code = int(response.status[:3])
mimetype = ''
if response.content_type:
mimetype = response.content_type.partition(";")[0]
else:
mimetype = response.options.default_media_type
return OpenAPIResponse(
data=response.body,
status_code=status_code,
mimetype=mimetype,
)

0
openapi_core/contrib/falcon/views.py Normal file
View file

1
requirements_dev.txt
View file

@ -2,6 +2,7 @@ mock==2.0.0
pytest==3.5.0
pytest-flake8
pytest-cov==2.5.1
falcon==2.0.0
flask
django==2.2.10; python_version>="3.0"
requests==2.22.0

1
setup.cfg
View file

@ -38,6 +38,7 @@ tests_require =
pytest
pytest-flake8
pytest-cov
falcon
flask
webob

51
tests/integration/contrib/falcon/conftest.py Normal file
View file

@ -0,0 +1,51 @@
from falcon import Request, Response, RequestOptions, ResponseOptions
from falcon.routing import DefaultRouter
from falcon.status_codes import HTTP_200
from falcon.testing import create_environ
import pytest
@pytest.fixture
def environ_factory():
def create_env(method, path, server_name):
return create_environ(
host=server_name,
path=path,
)
return create_env
@pytest.fixture
def router():
router = DefaultRouter()
router.add_route("/browse/{id:int}/", lambda x: x)
return router
@pytest.fixture
def request_factory(environ_factory, router):
server_name = 'localhost'
def create_request(
method, path, subdomain=None, query_string=None,
content_type='application/json'):
environ = environ_factory(method, path, server_name)
options = RequestOptions()
# return create_req(options=options, **environ)
req = Request(environ, options)
resource, method_map, params, req.uri_template = router.find(path, req)
return req
return create_request
@pytest.fixture
def response_factory(environ_factory):
def create_response(
data, status_code=200, content_type='application/json'):
options = ResponseOptions()
resp = Response(options)
resp.body = data
resp.content_type = content_type
resp.status = HTTP_200
return resp
return create_response

48
tests/integration/contrib/falcon/data/v3.0/falcon_factory.yaml Normal file
View file

@ -0,0 +1,48 @@
openapi: "3.0.0"
info:
title: Basic OpenAPI specification used with test_falcon.TestFalconOpenAPIIValidation
version: "0.1"
servers:
- url: 'http://localhost'
paths:
'/browse/{id}':
parameters:
- name: id
in: path
required: true
description: the ID of the resource to retrieve
schema:
type: integer
get:
responses:
200:
description: Return the resource.
content:
application/json:
schema:
type: object
required:
- data
properties:
data:
type: string
default:
description: Return errors.
content:
application/json:
schema:
type: object
required:
- errors
properties:
errors:
type: array
items:
type: object
properties:
title:
type: string
code:
type: string
message:
type: string

204
tests/integration/contrib/falcon/test_falcon_middlewares.py Normal file
View file

@ -0,0 +1,204 @@
from json import dumps
from falcon import API
from falcon.testing import TestClient
import pytest
from openapi_core.contrib.falcon.middlewares import FalconOpenAPIMiddleware
from openapi_core.shortcuts import create_spec
from openapi_core.validation.request.datatypes import RequestParameters
class TestFalconOpenAPIMiddleware(object):
view_response_callable = None
@pytest.fixture
def spec(self, factory):
specfile = 'contrib/falcon/data/v3.0/falcon_factory.yaml'
return create_spec(factory.spec_from_file(specfile))
@pytest.fixture
def middleware(self, spec):
return FalconOpenAPIMiddleware.from_spec(spec)
@pytest.fixture
def app(self, middleware):
return API(middleware=[middleware])
@pytest.yield_fixture
def client(self, app):
return TestClient(app)
@pytest.fixture
def view_response(self):
def view_response(*args, **kwargs):
return self.view_response_callable(*args, **kwargs)
return view_response
@pytest.fixture(autouse=True)
def details_view(self, app, view_response):
class BrowseDetailResource(object):
def on_get(self, *args, **kwargs):
return view_response(*args, **kwargs)
resource = BrowseDetailResource()
app.add_route("/browse/{id}", resource)
return resource
@pytest.fixture(autouse=True)
def list_view(self, app, view_response):
class BrowseListResource(object):
def on_get(self, *args, **kwargs):
return view_response(*args, **kwargs)
resource = BrowseListResource()
app.add_route("/browse", resource)
return resource
def test_invalid_content_type(self, client):
def view_response_callable(request, response, id):
from falcon.constants import MEDIA_HTML
from falcon.status_codes import HTTP_200
assert request.openapi
assert not request.openapi.errors
assert request.openapi.parameters == RequestParameters(path={
'id': 12,
})
response.content_type = MEDIA_HTML
response.status = HTTP_200
response.body = 'success'
self.view_response_callable = view_response_callable
headers = {'Content-Type': 'application/json'}
result = client.simulate_get(
'/browse/12', host='localhost', headers=headers)
assert result.json == {
'errors': [
{
'class': (
"<class 'openapi_core.schema.media_types.exceptions."
"InvalidContentType'>"
),
'status': 415,
'title': (
'Content for following mimetype not found: text/html'
)
}
]
}
def test_server_error(self, client):
headers = {'Content-Type': 'application/json'}
result = client.simulate_get(
'/browse/12', host='localhost', headers=headers, protocol='https')
expected_data = {
'errors': [
{
'class': (
"<class 'openapi_core.templating.paths.exceptions."
"ServerNotFound'>"
),
'status': 400,
'title': (
'Server not found for '
'https://localhost/browse/12'
),
}
]
}
assert result.status_code == 400
assert result.json == expected_data
def test_operation_error(self, client):
headers = {'Content-Type': 'application/json'}
result = client.simulate_post(
'/browse/12', host='localhost', headers=headers)
expected_data = {
'errors': [
{
'class': (
"<class 'openapi_core.templating.paths.exceptions."
"OperationNotFound'>"
),
'status': 405,
'title': (
'Operation post not found for '
'http://localhost/browse/12'
),
}
]
}
assert result.status_code == 405
assert result.json == expected_data
def test_path_error(self, client):
headers = {'Content-Type': 'application/json'}
result = client.simulate_get(
'/browse', host='localhost', headers=headers)
expected_data = {
'errors': [
{
'class': (
"<class 'openapi_core.templating.paths.exceptions."
"PathNotFound'>"
),
'status': 404,
'title': (
'Path not found for '
'http://localhost/browse'
),
}
]
}
assert result.status_code == 404
assert result.json == expected_data
def test_endpoint_error(self, client):
headers = {'Content-Type': 'application/json'}
result = client.simulate_get(
'/browse/invalidparameter', host='localhost', headers=headers)
expected_data = {
'errors': [
{
'class': (
"<class 'openapi_core.casting.schemas.exceptions."
"CastError'>"
),
'status': 400,
'title': (
"Failed to cast value invalidparameter to type integer"
)
}
]
}
assert result.json == expected_data
def test_valid(self, client):
def view_response_callable(request, response, id):
from falcon.constants import MEDIA_JSON
from falcon.status_codes import HTTP_200
assert request.openapi
assert not request.openapi.errors
assert request.openapi.parameters == RequestParameters(path={
'id': 12,
})
response.status = HTTP_200
response.content_type = MEDIA_JSON
response.body = dumps({
'data': 'data',
})
self.view_response_callable = view_response_callable
headers = {'Content-Type': 'application/json'}
result = client.simulate_get(
'/browse/12', host='localhost', headers=headers)
assert result.status_code == 200
assert result.json == {
'data': 'data',
}

34
tests/integration/contrib/falcon/test_falcon_validation.py Normal file
View file

@ -0,0 +1,34 @@
import pytest
from openapi_core.contrib.falcon.requests import FalconOpenAPIRequestFactory
from openapi_core.contrib.falcon.responses import FalconOpenAPIResponseFactory
from openapi_core.shortcuts import create_spec
from openapi_core.validation.request.validators import RequestValidator
from openapi_core.validation.response.validators import ResponseValidator
class TestFalconOpenAPIValidation(object):
@pytest.fixture
def spec(self, factory):
specfile = 'contrib/falcon/data/v3.0/falcon_factory.yaml'
return create_spec(factory.spec_from_file(specfile))
def test_response_validator_path_pattern(self,
spec,
request_factory,
response_factory):
validator = ResponseValidator(spec)
request = request_factory('GET', '/browse/12', subdomain='kb')
openapi_request = FalconOpenAPIRequestFactory.create(request)
response = response_factory('{"data": "data"}', status_code=200)
openapi_response = FalconOpenAPIResponseFactory.create(response)
result = validator.validate(openapi_request, openapi_response)
assert not result.errors
def test_request_validator_path_pattern(self, spec, request_factory):
validator = RequestValidator(spec)
request = request_factory('GET', '/browse/12', subdomain='kb')
openapi_request = FalconOpenAPIRequestFactory.create(request)
result = validator.validate(openapi_request)
assert not result.errors