servers with request validation

Author: p1c2u

openapi_core/exceptions.py

@@ -15,3 +15,7 @@ class MissingParameterError(OpenAPIMappingError):
 
 class InvalidContentTypeError(OpenAPIMappingError):
     pass
+
+
+class InvalidServerError(OpenAPIMappingError):
+    pass

openapi_core/servers.py

@@ -0,0 +1,76 @@
+from functools import lru_cache
+
+from six import iteritems
+
+
+class Server(object):
+
+    def __init__(self, url, variables=None):
+        self.url = url
+        self.variables = variables and dict(variables) or {}
+
+    @property
+    def default_url(self):
+        return self.get_url()
+
+    @property
+    def default_variables(self):
+        defaults = {}
+        for name, variable in iteritems(self.variables):
+            defaults[name] = variable.default
+        return defaults
+
+    def get_url(self, **variables):
+        if not variables:
+            variables = self.default_variables
+        return self.url.format(**variables)
+
+
+class ServerVariable(object):
+
+    def __init__(self, name, default, enum=None):
+        self.name = name
+        self.default = default
+        self.enum = enum and list(enum) or []
+
+
+class ServersGenerator(object):
+
+    def __init__(self, dereferencer):
+        self.dereferencer = dereferencer
+
+    def generate(self, servers_spec):
+        servers_deref = self.dereferencer.dereference(servers_spec)
+        for server_spec in servers_deref:
+            url = server_spec['url']
+            variables_spec = server_spec.get('variables', {})
+
+            variables = None
+            if variables_spec:
+                variables = self.variables_generator.generate(variables_spec)
+
+            yield Server(url, variables=variables)
+
+    @property
+    @lru_cache()
+    def variables_generator(self):
+        return ServerVariablesGenerator(self.dereferencer)
+
+
+class ServerVariablesGenerator(object):
+
+    def __init__(self, dereferencer):
+        self.dereferencer = dereferencer
+
+    def generate(self, variables_spec):
+        variables_deref = self.dereferencer.dereference(variables_spec)
+
+        if not variables_deref:
+            return [Server('/'), ]
+
+        for variable_name, variable_spec in iteritems(variables_deref):
+            default = variable_spec['default']
+            enum = variable_spec.get('enum')
+
+            variable = ServerVariable(variable_name, default, enum=enum)
+            yield variable_name, variable

openapi_core/specs.py

@@ -9,6 +9,7 @@
 from openapi_core.infos import InfoFactory
 from openapi_core.paths import PathsGenerator
 from openapi_core.schemas import SchemaRegistry
+from openapi_core.servers import ServersGenerator
 
 
 log = logging.getLogger(__name__)
@@ -26,8 +27,12 @@ def __init__(self, info, paths, servers=None, components=None):
     def __getitem__(self, path_name):
         return self.paths[path_name]
 
+    @property
+    def default_url(self):
+        return self.servers[0].default_url
+
     def get_server_url(self, index=0):
-        return self.servers[index]['url']
+        return self.servers[index].default_url
 
     def get_operation(self, path_pattern, http_method):
         return self.paths[path_pattern].operations[http_method]
@@ -59,14 +64,16 @@ def create(self, spec_dict, spec_url=''):
         spec_dict_deref = self.dereferencer.dereference(spec_dict)
 
         info_spec = spec_dict_deref.get('info', [])
-        servers = spec_dict_deref.get('servers', [])
+        servers_spec = spec_dict_deref.get('servers', [])
         paths = spec_dict_deref.get('paths', [])
         components_spec = spec_dict_deref.get('components', [])
 
         info = self.info_factory.create(info_spec)
+        servers = self.servers_generator.generate(servers_spec)
         paths = self.paths_generator.generate(paths)
         components = self.components_factory.create(components_spec)
-        spec = Spec(info, list(paths), servers=servers, components=components)
+        spec = Spec(
+            info, list(paths), servers=list(servers), components=components)
         return spec
 
     @property
@@ -79,6 +86,11 @@ def schemas_registry(self):
     def info_factory(self):
         return InfoFactory(self.dereferencer)
 
+    @property
+    @lru_cache()
+    def servers_generator(self):
+        return ServersGenerator(self.dereferencer)
+
     @property
     @lru_cache()
     def paths_generator(self):

openapi_core/wrappers.py

@@ -1,8 +1,10 @@
 """OpenAPI core wrappers module"""
 from six import iteritems
+from six.moves.urllib.parse import urljoin
 
 from openapi_core.exceptions import (
     OpenAPIMappingError, MissingParameterError, InvalidContentTypeError,
+    InvalidServerError,
 )
 
 SPEC_LOCATION_TO_REQUEST_LOCATION_MAPPING = {
@@ -32,13 +34,32 @@ def validate_location(cls, location):
                 "Unknown parameter location: {0}".format(str(location)))
 
 
-class RequestParametersFactory(object):
+class BaseRequestFactory(object):
+
+    def get_operation(self, request, spec):
+        server = self._get_server(request, spec)
+
+        operation_pattern = request.full_url_pattern.replace(
+            server.default_url, '')
+
+        return spec.get_operation(operation_pattern, request.method)
+
+    def _get_server(self, request, spec):
+        for server in spec.servers:
+            if server.default_url in request.full_url_pattern:
+                return server
+
+        raise InvalidServerError(
+            "Invalid request server {0}".format(request.full_url_pattern))
+
+
+class RequestParametersFactory(BaseRequestFactory):
 
     def __init__(self, attr_mapping=SPEC_LOCATION_TO_REQUEST_LOCATION_MAPPING):
         self.attr_mapping = attr_mapping
 
     def create(self, request, spec):
-        operation = spec.get_operation(request.path_pattern, request.method)
+        operation = self.get_operation(request, spec)
 
         params = RequestParameters()
         for param_name, param in iteritems(operation.parameters):
@@ -65,10 +86,10 @@ def _unmarshal_param(self, request, param):
         return param.unmarshal(raw_value)
 
 
-class RequestBodyFactory(object):
+class RequestBodyFactory(BaseRequestFactory):
 
     def create(self, request, spec):
-        operation = spec.get_operation(request.path_pattern, request.method)
+        operation = self.get_operation(request, spec)
 
         try:
             media_type = operation.request_body[request.content_type]
@@ -78,9 +99,13 @@ def create(self, request, spec):
 
         return media_type.unmarshal(request.data)
 
+    def _get_operation(self, request, spec):
+        return spec.get_operation(request.path_pattern, request.method)
+
 
 class BaseOpenAPIRequest(object):
 
+    host_url = NotImplemented
     path = NotImplemented
     path_pattern = NotImplemented
     method = NotImplemented
@@ -94,6 +119,10 @@ class BaseOpenAPIRequest(object):
 
     content_type = NotImplemented
 
+    @property
+    def full_url_pattern(self):
+        return urljoin(self.host_url, self.path_pattern)
+
     def get_parameters(self, spec):
         return RequestParametersFactory().create(self, spec)
 

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

@@ -5,7 +5,13 @@ info:
   license:
     name: MIT
 servers:
-  - url: http://petstore.swagger.io/v1
+  - url: http://petstore.swagger.io/{version}
+    variables:
+      version:
+        enum:
+          - v1
+          - v2
+        default: v1
 paths:
   /pets:
     get: