Responses and headers objects

Author: p1c2u

openapi_core/exceptions.py

@@ -69,5 +69,9 @@ class InvalidContentType(OpenAPIBodyError):
     pass
 
 
+class InvalidResponse(OpenAPIMappingError):
+    pass
+
+
 class InvalidValue(OpenAPIMappingError):
     pass

openapi_core/operations.py

@@ -5,8 +5,10 @@
 
 from six import iteritems
 
+from openapi_core.exceptions import InvalidResponse
 from openapi_core.parameters import ParametersGenerator
 from openapi_core.request_bodies import RequestBodyFactory
+from openapi_core.responses import ResponsesGenerator
 
 log = logging.getLogger(__name__)
 
@@ -15,10 +17,11 @@ class Operation(object):
     """Represents an OpenAPI Operation."""
 
     def __init__(
-            self, http_method, path_name, parameters, request_body=None,
-            deprecated=False, operation_id=None):
+            self, http_method, path_name, responses, parameters,
+            request_body=None, deprecated=False, operation_id=None):
         self.http_method = http_method
         self.path_name = path_name
+        self.responses = dict(responses)
         self.parameters = dict(parameters)
         self.request_body = request_body
         self.deprecated = deprecated
@@ -27,6 +30,16 @@ def __init__(
     def __getitem__(self, name):
         return self.parameters[name]
 
+    def get_response(self, http_status='default'):
+        try:
+            return self.responses[http_status]
+        except KeyError:
+            if 'default' not in self.responses:
+                raise InvalidResponse(
+                    "Unknown response http status {0}".format(http_status))
+
+            return self.responses['default']
+
 
 class OperationsGenerator(object):
     """Represents an OpenAPI Operation in a service."""
@@ -42,9 +55,12 @@ def generate(self, path_name, path):
                 continue
 
             operation_deref = self.dereferencer.dereference(operation)
+            responses_spec = operation_deref['responses']
+            responses = self.responses_generator.generate(responses_spec)
             deprecated = operation_deref.get('deprecated', False)
             parameters_list = operation_deref.get('parameters', [])
-            parameters = self.parameters_generator.generate(parameters_list)
+            parameters = self.parameters_generator.generate_from_list(
+                parameters_list)
 
             request_body = None
             if 'requestBody' in operation_deref:
@@ -55,11 +71,16 @@ def generate(self, path_name, path):
             yield (
                 http_method,
                 Operation(
-                    http_method, path_name, list(parameters),
+                    http_method, path_name, responses, list(parameters),
                     request_body=request_body, deprecated=deprecated,
                 ),
             )
 
+    @property
+    @lru_cache()
+    def responses_generator(self):
+        return ResponsesGenerator(self.dereferencer, self.schemas_registry)
+
     @property
     @lru_cache()
     def parameters_generator(self):

openapi_core/parameters.py

@@ -2,6 +2,8 @@
 import logging
 import warnings
 
+from six import iteritems
+
 from openapi_core.exceptions import (
     EmptyValue, InvalidValueType, InvalidParameterValue,
 )
@@ -54,10 +56,36 @@ def __init__(self, dereferencer, schemas_registry):
         self.dereferencer = dereferencer
         self.schemas_registry = schemas_registry
 
-    def generate(self, paramters):
-        for parameter in paramters:
+    def generate(self, parameters):
+        for parameter_name, parameter in iteritems(parameters):
+            parameter_deref = self.dereferencer.dereference(parameter)
+
+            parameter_in = parameter_deref.get('in', 'header')
+
+            allow_empty_value = parameter_deref.get('allowEmptyValue')
+            required = parameter_deref.get('required', False)
+
+            schema_spec = parameter_deref.get('schema', None)
+            schema = None
+            if schema_spec:
+                schema, _ = self.schemas_registry.get_or_create(schema_spec)
+
+            yield (
+                parameter_name,
+                Parameter(
+                    parameter_name, parameter_in,
+                    schema=schema, required=required,
+                    allow_empty_value=allow_empty_value,
+                ),
+            )
+
+    def generate_from_list(self, parameters_list):
+        for parameter in parameters_list:
             parameter_deref = self.dereferencer.dereference(parameter)
 
+            parameter_name = parameter_deref['name']
+            parameter_in = parameter_deref.get('in', 'header')
+
             allow_empty_value = parameter_deref.get('allowEmptyValue')
             required = parameter_deref.get('required', False)
 
@@ -67,9 +95,9 @@ def generate(self, paramters):
                 schema, _ = self.schemas_registry.get_or_create(schema_spec)
 
             yield (
-                parameter_deref['name'],
+                parameter_name,
                 Parameter(
-                    parameter_deref['name'], parameter_deref['in'],
+                    parameter_name, parameter_in,
                     schema=schema, required=required,
                     allow_empty_value=allow_empty_value,
                 ),

openapi_core/responses.py

@@ -0,0 +1,54 @@
+"""OpenAPI core responses module"""
+from functools import lru_cache
+
+from six import iteritems
+
+from openapi_core.media_types import MediaTypeGenerator
+from openapi_core.parameters import ParametersGenerator
+
+
+class Response(object):
+
+    def __init__(
+            self, http_status, description, headers=None, content=None,
+            links=None):
+        self.http_status = http_status
+        self.description = description
+        self.headers = headers and dict(headers) or {}
+        self.content = content and dict(content) or {}
+        self.links = links and dict(links) or {}
+
+
+class ResponsesGenerator(object):
+
+    def __init__(self, dereferencer, schemas_registry):
+        self.dereferencer = dereferencer
+        self.schemas_registry = schemas_registry
+
+    def generate(self, responses):
+        for http_status, response in iteritems(responses):
+            description = response['description']
+            headers = response.get('headers')
+            content = response.get('content')
+
+            media_types = None
+            if content:
+                media_types = self.media_types_generator.generate(content)
+
+            parameters = None
+            if headers:
+                parameters = self.parameters_generator.generate(headers)
+
+            yield http_status, Response(
+                http_status, description,
+                content=media_types, headers=parameters)
+
+    @property
+    @lru_cache()
+    def media_types_generator(self):
+        return MediaTypeGenerator(self.dereferencer, self.schemas_registry)
+
+    @property
+    @lru_cache()
+    def parameters_generator(self):
+        return ParametersGenerator(self.dereferencer, self.schemas_registry)

tests/integration/test_petstore.py

@@ -9,8 +9,10 @@
 )
 from openapi_core.media_types import MediaType
 from openapi_core.operations import Operation
+from openapi_core.parameters import Parameter
 from openapi_core.paths import Path
 from openapi_core.request_bodies import RequestBody
+from openapi_core.responses import Response
 from openapi_core.schemas import Schema
 from openapi_core.servers import Server, ServerVariable
 from openapi_core.shortcuts import create_spec
@@ -59,6 +61,60 @@ def test_spec(self, spec, spec_dict):
                 assert operation.http_method == http_method
 
                 operation_spec = spec_dict['paths'][path_name][http_method]
+
+                responses_spec = operation_spec.get('responses')
+
+                for http_status, response in iteritems(operation.responses):
+                    assert type(response) == Response
+                    assert response.http_status == http_status
+
+                    response_spec = responses_spec[http_status]
+                    description_spec = response_spec['description']
+
+                    assert response.description == description_spec
+
+                    for mimetype, media_type in iteritems(response.content):
+                        assert type(media_type) == MediaType
+                        assert media_type.mimetype == mimetype
+
+                        content_spec = response_spec['content'][mimetype]
+                        schema_spec = content_spec.get('schema')
+                        assert bool(schema_spec) == bool(media_type.schema)
+
+                        if not schema_spec:
+                            continue
+
+                        # @todo: test with defererence
+                        if '$ref' in schema_spec:
+                            continue
+
+                        assert type(media_type.schema) == Schema
+                        assert media_type.schema.type == schema_spec['type']
+                        assert media_type.schema.required == schema_spec.get(
+                            'required', False)
+
+                    for parameter_name, parameter in iteritems(
+                            response.headers):
+                        assert type(parameter) == Parameter
+                        assert parameter.name == parameter_name
+
+                        headers_spec = response_spec['headers']
+                        parameter_spec = headers_spec[parameter_name]
+                        schema_spec = parameter_spec.get('schema')
+                        assert bool(schema_spec) == bool(parameter.schema)
+
+                        if not schema_spec:
+                            continue
+
+                        # @todo: test with defererence
+                        if '$ref' in schema_spec:
+                            continue
+
+                        assert type(parameter.schema) == Schema
+                        assert parameter.schema.type == schema_spec['type']
+                        assert parameter.schema.required == schema_spec.get(
+                            'required', False)
+
                 request_body_spec = operation_spec.get('requestBody')
 
                 assert bool(request_body_spec) == bool(operation.request_body)