Flask OpenAPI view

openapi_core/contrib/flask/decorators.py

@@ -0,0 +1,46 @@
+"""OpenAPI core contrib flask decorators module"""
+from openapi_core.contrib.flask.handlers import FlaskOpenAPIErrorsHandler
+from openapi_core.contrib.flask.providers import FlaskRequestProvider
+from openapi_core.contrib.flask.requests import FlaskOpenAPIRequestFactory
+from openapi_core.contrib.flask.responses import FlaskOpenAPIResponseFactory
+from openapi_core.validation.decorators import OpenAPIDecorator
+from openapi_core.validation.request.validators import RequestValidator
+from openapi_core.validation.response.validators import ResponseValidator
+
+
+class FlaskOpenAPIViewDecorator(OpenAPIDecorator):
+
+    def __init__(
+            self,
+            request_validator,
+            response_validator,
+            request_factory=FlaskOpenAPIRequestFactory,
+            response_factory=FlaskOpenAPIResponseFactory,
+            request_provider=FlaskRequestProvider,
+            openapi_errors_handler=FlaskOpenAPIErrorsHandler,
+    ):
+        super(FlaskOpenAPIViewDecorator, self).__init__(
+            request_validator, response_validator,
+            request_factory, response_factory,
+            request_provider, openapi_errors_handler,
+        )
+
+    @classmethod
+    def from_spec(
+            cls,
+            spec,
+            request_factory=FlaskOpenAPIRequestFactory,
+            response_factory=FlaskOpenAPIResponseFactory,
+            request_provider=FlaskRequestProvider,
+            openapi_errors_handler=FlaskOpenAPIErrorsHandler,
+    ):
+        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,
+            request_provider=request_provider,
+            openapi_errors_handler=openapi_errors_handler,
+        )

openapi_core/contrib/flask/handlers.py

@@ -0,0 +1,41 @@
+"""OpenAPI core contrib flask handlers module"""
+from flask.globals import current_app
+from flask.json import dumps
+
+from openapi_core.schema.media_types.exceptions import InvalidContentType
+from openapi_core.schema.servers.exceptions import InvalidServer
+
+
+class FlaskOpenAPIErrorsHandler(object):
+
+    OPENAPI_ERROR_STATUS = {
+        InvalidServer: 500,
+        InvalidContentType: 415,
+    }
+
+    @classmethod
+    def handle(cls, errors):
+        data_errors = [
+            cls.format_openapi_error(err)
+            for err in errors
+        ]
+        data = {
+            'errors': data_errors,
+        }
+        status = max(
+            range(len(data_errors)),
+            key=lambda idx: data_errors[idx]['status'],
+        )
+        return current_app.response_class(
+            dumps(data),
+            status=status,
+            mimetype='application/json'
+        )
+
+    @classmethod
+    def format_openapi_error(cls, error):
+        return {
+            'title': str(error),
+            'status': cls.OPENAPI_ERROR_STATUS.get(error.__class__, 400),
+            'class': str(type(error)),
+        }

openapi_core/contrib/flask/providers.py

@@ -0,0 +1,9 @@
+"""OpenAPI core contrib flask providers module"""
+from flask.globals import request
+
+
+class FlaskRequestProvider(object):
+
+    @classmethod
+    def provide(self, *args, **kwargs):
+        return request

openapi_core/contrib/flask/views.py

@@ -0,0 +1,42 @@
+"""OpenAPI core contrib flask views module"""
+from flask.views import MethodView
+
+from openapi_core.contrib.flask.decorators import FlaskOpenAPIViewDecorator
+from openapi_core.validation.request.validators import RequestValidator
+from openapi_core.validation.response.validators import ResponseValidator
+
+
+class FlaskOpenAPIView(MethodView):
+    """Brings OpenAPI specification validation and unmarshalling for views."""
+
+    def __init__(self, request_validator, response_validator):
+        super(MethodView, self).__init__()
+        self.request_validator = request_validator
+        self.response_validator = response_validator
+
+    def dispatch_request(self, *args, **kwargs):
+        decorator = FlaskOpenAPIViewDecorator(
+            request_validator=self.request_validator,
+            response_validator=self.response_validator,
+            openapi_errors_handler=self.handle_openapi_errors,
+        )
+        return decorator(super(FlaskOpenAPIView, self).dispatch_request)(
+            *args, **kwargs)
+
+    def handle_openapi_errors(self, errors):
+        """Handles OpenAPI request/response errors.
+
+        Should return response object::
+
+            class MyView(FlaskOpenAPIView):
+
+                def handle_openapi_errors(self, errors):
+                    return jsonify({'errors': errors})
+        """
+        raise NotImplementedError
+
+    @classmethod
+    def from_spec(cls, spec):
+        request_validator = RequestValidator(spec)
+        response_validator = ResponseValidator(spec)
+        return cls(request_validator, response_validator)

openapi_core/validation/decorators.py

@@ -0,0 +1,51 @@
+"""OpenAPI core validation decorators module"""
+from functools import wraps
+
+from openapi_core.validation.processors import OpenAPIProcessor
+
+
+class OpenAPIDecorator(OpenAPIProcessor):
+
+    def __init__(
+            self,
+            request_validator,
+            response_validator,
+            request_factory,
+            response_factory,
+            request_provider,
+            openapi_errors_handler,
+    ):
+        super(OpenAPIDecorator, self).__init__(
+            request_validator, response_validator)
+        self.request_factory = request_factory
+        self.response_factory = response_factory
+        self.request_provider = request_provider
+        self.openapi_errors_handler = openapi_errors_handler
+
+    def __call__(self, view):
+        @wraps(view)
+        def decorated(*args, **kwargs):
+            request = self._get_request(*args, **kwargs)
+            openapi_request = self._get_openapi_request(request)
+            errors = self.process_request(openapi_request)
+            if errors:
+                return self._handle_openapi_errors(errors)
+            response = view(*args, **kwargs)
+            openapi_response = self._get_openapi_response(response)
+            errors = self.process_response(openapi_request, openapi_response)
+            if errors:
+                return self._handle_openapi_errors(errors)
+            return response
+        return decorated
+
+    def _get_request(self, *args, **kwargs):
+        return self.request_provider.provide(*args, **kwargs)
+
+    def _handle_openapi_errors(self, errors):
+        return self.openapi_errors_handler.handle(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)

openapi_core/validation/processors.py

@@ -0,0 +1,31 @@
+"""OpenAPI core validation processors module"""
+from openapi_core.schema.servers.exceptions import InvalidServer
+from openapi_core.schema.exceptions import OpenAPIMappingError
+
+
+class OpenAPIProcessor(object):
+
+    def __init__(self, request_validator, response_validator):
+        self.request_validator = request_validator
+        self.response_validator = response_validator
+
+    def process_request(self, request):
+        request_result = self.request_validator.validate(request)
+        try:
+            request_result.raise_for_errors()
+        # return instantly on server error
+        except InvalidServer as exc:
+            return [exc, ]
+        except OpenAPIMappingError:
+            return request_result.errors
+        else:
+            return
+
+    def process_response(self, request, response):
+        response_result = self.response_validator.validate(request, response)
+        try:
+            response_result.raise_for_errors()
+        except OpenAPIMappingError:
+            return response_result.errors
+        else:
+            return

tests/integration/conftest.py

@@ -1,5 +1,6 @@
 from os import path
 
+from openapi_spec_validator.schemas import read_yaml_file
 import pytest
 from six.moves.urllib import request
 from yaml import safe_load
@@ -8,8 +9,7 @@ from yaml import safe_load
 def spec_from_file(spec_file):
     directory = path.abspath(path.dirname(__file__))
     path_full = path.join(directory, spec_file)
-    with open(path_full) as fh:
-        return safe_load(fh)
+    return read_yaml_file(path_full)
 
 
 def spec_from_url(spec_url):

tests/integration/contrib/flask/conftest.py

@@ -0,0 +1,49 @@
+from flask.wrappers import Request, Response
+import pytest
+from werkzeug.routing import Map, Rule, Subdomain
+from werkzeug.test import create_environ
+
+
+@pytest.fixture
+def environ_factory():
+    return create_environ
+
+
+@pytest.fixture
+def map():
+    return Map([
+        # Static URLs
+        Rule('/', endpoint='static/index'),
+        Rule('/about', endpoint='static/about'),
+        Rule('/help', endpoint='static/help'),
+        # Knowledge Base
+        Subdomain('kb', [
+            Rule('/', endpoint='kb/index'),
+            Rule('/browse/', endpoint='kb/browse'),
+            Rule('/browse/<int:id>/', endpoint='kb/browse'),
+            Rule('/browse/<int:id>/<int:page>', endpoint='kb/browse')
+        ])
+    ], default_subdomain='www')
+
+
+@pytest.fixture
+def request_factory(map, environ_factory):
+    server_name = 'localhost'
+
+    def create_request(method, path, subdomain=None, query_string=None):
+        environ = environ_factory(query_string=query_string)
+        req = Request(environ)
+        urls = map.bind_to_environ(
+            environ, server_name=server_name, subdomain=subdomain)
+        req.url_rule, req.view_args = urls.match(
+            path, method, return_rule=True)
+        return req
+    return create_request
+
+
+@pytest.fixture
+def response_factory():
+    def create_response(
+            data, status_code=200, content_type='application/json'):
+        return Response(data, status=status_code, content_type=content_type)
+    return create_response

tests/integration/contrib/flask/data/v3.0/flask_factory.yaml

@@ -0,0 +1,48 @@
+openapi: "3.0.0"
+info:
+  title: Basic OpenAPI specification used with test_flask.TestFlaskOpenAPIIValidation
+  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

tests/integration/contrib/flask/test_flask_decorator.py

@@ -0,0 +1,111 @@
+from flask import Flask, make_response, jsonify
+import pytest
+
+from openapi_core.contrib.flask.decorators import FlaskOpenAPIViewDecorator
+from openapi_core.shortcuts import create_spec
+
+
+class TestFlaskOpenAPIDecorator(object):
+
+    view_response = None
+
+    @pytest.fixture
+    def spec(self, factory):
+        specfile = 'contrib/flask/data/v3.0/flask_factory.yaml'
+        return create_spec(factory.spec_from_file(specfile))
+
+    @pytest.fixture
+    def decorator(self, spec):
+        return FlaskOpenAPIViewDecorator.from_spec(spec)
+
+    @pytest.fixture
+    def app(self):
+        app = Flask("__main__")
+        app.config['DEBUG'] = True
+        app.config['TESTING'] = True
+        return app
+
+    @pytest.yield_fixture
+    def client(self, app):
+        with app.test_client() as client:
+            with app.app_context():
+                yield client
+
+    @pytest.fixture(autouse=True)
+    def view(self, app, decorator):
+        @app.route("/browse/<id>/")
+        @decorator
+        def browse_details(id):
+            return self.view_response
+        return browse_details
+
+    def test_invalid_content_type(self, client):
+        self.view_response = make_response('success', 200)
+
+        result = client.get('/browse/12/')
+
+        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):
+        result = client.get('/browse/12/', base_url='https://localhost')
+
+        expected_data = {
+            'errors': [
+                {
+                    'class': (
+                        "<class 'openapi_core.schema.servers.exceptions."
+                        "InvalidServer'>"
+                    ),
+                    'status': 500,
+                    'title': (
+                        'Invalid request server '
+                        'https://localhost/browse/{id}/'
+                    ),
+                }
+            ]
+        }
+        assert result.json == expected_data
+
+    def test_endpoint_error(self, client):
+        result = client.get('/browse/invalidparameter/')
+
+        expected_data = {
+            'errors': [
+                {
+                    'class': (
+                        "<class 'openapi_core.schema.parameters."
+                        "exceptions.InvalidParameterValue'>"
+                    ),
+                    'status': 400,
+                    'title': (
+                        'Invalid parameter value for `id`: '
+                        'Failed to cast value invalidparameter to type '
+                        'SchemaType.INTEGER'
+                    )
+                }
+            ]
+        }
+        assert result.json == expected_data
+
+    def test_valid(self, client):
+        self.view_response = jsonify(data='data')
+
+        result = client.get('/browse/12/')
+
+        assert result.status_code == 200
+        assert result.json == {
+            'data': 'data',
+        }

tests/integration/contrib/flask/test_flask_requests.py

@@ -0,0 +1,76 @@
+from werkzeug.datastructures import EnvironHeaders, ImmutableMultiDict
+
+from openapi_core.contrib.flask import FlaskOpenAPIRequest
+from openapi_core.validation.request.datatypes import RequestParameters
+
+
+class TestFlaskOpenAPIRequest(object):
+
+    def test_simple(self, request_factory, request):
+        request = request_factory('GET', '/', subdomain='www')
+
+        openapi_request = FlaskOpenAPIRequest(request)
+
+        path = {}
+        query = ImmutableMultiDict([])
+        headers = EnvironHeaders(request.environ)
+        cookies = {}
+        assert openapi_request.parameters == RequestParameters(
+            path=path,
+            query=query,
+            header=headers,
+            cookie=cookies,
+        )
+        assert openapi_request.host_url == request.host_url
+        assert openapi_request.path == request.path
+        assert openapi_request.method == request.method.lower()
+        assert openapi_request.path_pattern == request.path
+        assert openapi_request.body == request.data
+        assert openapi_request.mimetype == request.mimetype
+
+    def test_multiple_values(self, request_factory, request):
+        request = request_factory(
+            'GET', '/', subdomain='www', query_string='a=b&a=c')
+
+        openapi_request = FlaskOpenAPIRequest(request)
+
+        path = {}
+        query = ImmutableMultiDict([
+            ('a', 'b'), ('a', 'c'),
+        ])
+        headers = EnvironHeaders(request.environ)
+        cookies = {}
+        assert openapi_request.parameters == RequestParameters(
+            path=path,
+            query=query,
+            header=headers,
+            cookie=cookies,
+        )
+        assert openapi_request.host_url == request.host_url
+        assert openapi_request.path == request.path
+        assert openapi_request.method == request.method.lower()
+        assert openapi_request.path_pattern == request.path
+        assert openapi_request.body == request.data
+        assert openapi_request.mimetype == request.mimetype
+
+    def test_url_rule(self, request_factory, request):
+        request = request_factory('GET', '/browse/12/', subdomain='kb')
+
+        openapi_request = FlaskOpenAPIRequest(request)
+
+        path = {'id': 12}
+        query = ImmutableMultiDict([])
+        headers = EnvironHeaders(request.environ)
+        cookies = {}
+        assert openapi_request.parameters == RequestParameters(
+            path=path,
+            query=query,
+            header=headers,
+            cookie=cookies,
+        )
+        assert openapi_request.host_url == request.host_url
+        assert openapi_request.path == request.path
+        assert openapi_request.method == request.method.lower()
+        assert openapi_request.path_pattern == '/browse/{id}/'
+        assert openapi_request.body == request.data
+        assert openapi_request.mimetype == request.mimetype

tests/integration/contrib/flask/test_flask_responses.py

@@ -0,0 +1,13 @@
+from openapi_core.contrib.flask import FlaskOpenAPIResponse
+
+
+class TestFlaskOpenAPIResponse(object):
+
+    def test_invalid_server(self, response_factory):
+        response = response_factory('Not Found', status_code=404)
+
+        openapi_response = FlaskOpenAPIResponse(response)
+
+        assert openapi_response.data == response.data
+        assert openapi_response.status_code == response._status_code
+        assert openapi_response.mimetype == response.mimetype

tests/integration/contrib/flask/test_flask_validation.py

@@ -0,0 +1,35 @@
+import pytest
+
+from openapi_core.contrib.flask import (
+    FlaskOpenAPIRequest, FlaskOpenAPIResponse,
+)
+from openapi_core.shortcuts import create_spec
+from openapi_core.validation.request.validators import RequestValidator
+from openapi_core.validation.response.validators import ResponseValidator
+
+
+class TestFlaskOpenAPIValidation(object):
+
+    @pytest.fixture
+    def flask_spec(self, factory):
+        specfile = 'contrib/flask/data/v3.0/flask_factory.yaml'
+        return create_spec(factory.spec_from_file(specfile))
+
+    def test_response_validator_path_pattern(self,
+                                             flask_spec,
+                                             request_factory,
+                                             response_factory):
+        validator = ResponseValidator(flask_spec)
+        request = request_factory('GET', '/browse/12/', subdomain='kb')
+        openapi_request = FlaskOpenAPIRequest(request)
+        response = response_factory('{"data": "data"}', status_code=200)
+        openapi_response = FlaskOpenAPIResponse(response)
+        result = validator.validate(openapi_request, openapi_response)
+        assert not result.errors
+
+    def test_request_validator_path_pattern(self, flask_spec, request_factory):
+        validator = RequestValidator(flask_spec)
+        request = request_factory('GET', '/browse/12/', subdomain='kb')
+        openapi_request = FlaskOpenAPIRequest(request)
+        result = validator.validate(openapi_request)
+        assert not result.errors

tests/integration/contrib/test_flask.py

@@ -1,168 +0,0 @@
-from flask.wrappers import Request, Response
-import pytest
-from werkzeug.datastructures import EnvironHeaders, ImmutableMultiDict
-from werkzeug.routing import Map, Rule, Subdomain
-from werkzeug.test import create_environ
-
-from openapi_core.contrib.flask import (
-    FlaskOpenAPIRequest, FlaskOpenAPIResponse,
-)
-from openapi_core.shortcuts import create_spec
-from openapi_core.validation.request.datatypes import RequestParameters
-from openapi_core.validation.request.validators import RequestValidator
-from openapi_core.validation.response.validators import ResponseValidator
-
-
-@pytest.fixture
-def environ_factory():
-    return create_environ
-
-
-@pytest.fixture
-def map():
-    return Map([
-        # Static URLs
-        Rule('/', endpoint='static/index'),
-        Rule('/about', endpoint='static/about'),
-        Rule('/help', endpoint='static/help'),
-        # Knowledge Base
-        Subdomain('kb', [
-            Rule('/', endpoint='kb/index'),
-            Rule('/browse/', endpoint='kb/browse'),
-            Rule('/browse/<int:id>/', endpoint='kb/browse'),
-            Rule('/browse/<int:id>/<int:page>', endpoint='kb/browse')
-        ])
-    ], default_subdomain='www')
-
-
-@pytest.fixture
-def request_factory(map, environ_factory):
-    server_name = 'localhost'
-
-    def create_request(method, path, subdomain=None, query_string=None):
-        environ = environ_factory(query_string=query_string)
-        req = Request(environ)
-        urls = map.bind_to_environ(
-            environ, server_name=server_name, subdomain=subdomain)
-        req.url_rule, req.view_args = urls.match(
-            path, method, return_rule=True)
-        return req
-    return create_request
-
-
-@pytest.fixture
-def response_factory():
-    def create_response(data, status_code=200):
-        return Response(data, status=status_code)
-    return create_response
-
-
-class TestFlaskOpenAPIRequest(object):
-
-    def test_simple(self, request_factory, request):
-        request = request_factory('GET', '/', subdomain='www')
-
-        openapi_request = FlaskOpenAPIRequest(request)
-
-        path = {}
-        query = ImmutableMultiDict([])
-        headers = EnvironHeaders(request.environ)
-        cookies = {}
-        assert openapi_request.parameters == RequestParameters(
-            path=path,
-            query=query,
-            header=headers,
-            cookie=cookies,
-        )
-        assert openapi_request.host_url == request.host_url
-        assert openapi_request.path == request.path
-        assert openapi_request.method == request.method.lower()
-        assert openapi_request.path_pattern == request.path
-        assert openapi_request.body == request.data
-        assert openapi_request.mimetype == request.mimetype
-
-    def test_multiple_values(self, request_factory, request):
-        request = request_factory(
-            'GET', '/', subdomain='www', query_string='a=b&a=c')
-
-        openapi_request = FlaskOpenAPIRequest(request)
-
-        path = {}
-        query = ImmutableMultiDict([
-            ('a', 'b'), ('a', 'c'),
-        ])
-        headers = EnvironHeaders(request.environ)
-        cookies = {}
-        assert openapi_request.parameters == RequestParameters(
-            path=path,
-            query=query,
-            header=headers,
-            cookie=cookies,
-        )
-        assert openapi_request.host_url == request.host_url
-        assert openapi_request.path == request.path
-        assert openapi_request.method == request.method.lower()
-        assert openapi_request.path_pattern == request.path
-        assert openapi_request.body == request.data
-        assert openapi_request.mimetype == request.mimetype
-
-    def test_url_rule(self, request_factory, request):
-        request = request_factory('GET', '/browse/12/', subdomain='kb')
-
-        openapi_request = FlaskOpenAPIRequest(request)
-
-        path = {'id': 12}
-        query = ImmutableMultiDict([])
-        headers = EnvironHeaders(request.environ)
-        cookies = {}
-        assert openapi_request.parameters == RequestParameters(
-            path=path,
-            query=query,
-            header=headers,
-            cookie=cookies,
-        )
-        assert openapi_request.host_url == request.host_url
-        assert openapi_request.path == request.path
-        assert openapi_request.method == request.method.lower()
-        assert openapi_request.path_pattern == '/browse/{id}/'
-        assert openapi_request.body == request.data
-        assert openapi_request.mimetype == request.mimetype
-
-
-class TestFlaskOpenAPIResponse(object):
-
-    def test_invalid_server(self, response_factory):
-        response = response_factory('Not Found', status_code=404)
-
-        openapi_response = FlaskOpenAPIResponse(response)
-
-        assert openapi_response.data == response.data
-        assert openapi_response.status_code == response._status_code
-        assert openapi_response.mimetype == response.mimetype
-
-
-class TestFlaskOpenAPIValidation(object):
-
-    @pytest.fixture
-    def flask_spec(self, factory):
-        specfile = 'data/v3.0/flask_factory.yaml'
-        return create_spec(factory.spec_from_file(specfile))
-
-    def test_response_validator_path_pattern(self,
-                                             flask_spec,
-                                             request_factory,
-                                             response_factory):
-        validator = ResponseValidator(flask_spec)
-        request = request_factory('GET', '/browse/12/', subdomain='kb')
-        openapi_request = FlaskOpenAPIRequest(request)
-        response = response_factory('Some item', status_code=200)
-        openapi_response = FlaskOpenAPIResponse(response)
-        result = validator.validate(openapi_request, openapi_response)
-        assert not result.errors
-
-    def test_request_validator_path_pattern(self, flask_spec, request_factory):
-        validator = RequestValidator(flask_spec)
-        request = request_factory('GET', '/browse/12/', subdomain='kb')
-        openapi_request = FlaskOpenAPIRequest(request)
-        result = validator.validate(openapi_request)
-        assert not result.errors

tests/integration/data/v3.0/flask_factory.yaml

@@ -1,19 +0,0 @@
-openapi: "3.0.0"
-info:
-  title: Basic OpenAPI specification used with test_flask.TestFlaskOpenAPIIValidation
-  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:
-        default:
-          description: Return the resource.