Flask OpenAPI view

Author: p1c2u

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

openapi_core/wrappers/tastypie.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 @@
 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/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/conftest.py

@@ -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