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

Merge pull request #318 from p1c2u/feature/spec-path

Browse source 
Spec path
This commit is contained in:
A 2021-05-02 23:05:06 +01:00 committed by GitHub
commit 05fd27f04a
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
125 changed files with 1545 additions and 2897 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
openapi_core/casting/schemas/casters.py
View file

@ -1,5 +1,7 @@
from __future__ import division
from openapi_core.casting.schemas.exceptions import CastError
from openapi_core.schema.schemas.types import NoValue
from openapi_core.types import NoValue
class PrimitiveCaster(object):
@ -14,7 +16,7 @@ class PrimitiveCaster(object):
try:
return self.caster_callable(value)
except (ValueError, TypeError):
raise CastError(value, self.schema.type.value)
raise CastError(value, self.schema['type'])
class DummyCaster(object):
@ -31,7 +33,7 @@ class ArrayCaster(object):
@property
def items_caster(self):
return self.casters_factory.create(self.schema.items)
return self.casters_factory.create(self.schema / 'items')
def __call__(self, value):
if value in (None, NoValue):

23
openapi_core/casting/schemas/factories.py
View file

@ -1,5 +1,3 @@
from openapi_core.schema.schemas.enums import SchemaType
from openapi_core.casting.schemas.casters import (
PrimitiveCaster, DummyCaster, ArrayCaster
)
@ -9,23 +7,24 @@ from openapi_core.casting.schemas.util import forcebool
class SchemaCastersFactory(object):
DUMMY_CASTERS = [
SchemaType.STRING, SchemaType.OBJECT, SchemaType.ANY,
'string', 'object', 'any',
]
PRIMITIVE_CASTERS = {
SchemaType.INTEGER: int,
SchemaType.NUMBER: float,
SchemaType.BOOLEAN: forcebool,
'integer': int,
'number': float,
'boolean': forcebool,
}
COMPLEX_CASTERS = {
SchemaType.ARRAY: ArrayCaster,
'array': ArrayCaster,
}
def create(self, schema):
if schema.type in self.DUMMY_CASTERS:
schema_type = schema.getkey('type', 'any')
if schema_type in self.DUMMY_CASTERS:
return DummyCaster()
elif schema.type in self.PRIMITIVE_CASTERS:
caster_callable = self.PRIMITIVE_CASTERS[schema.type]
elif schema_type in self.PRIMITIVE_CASTERS:
caster_callable = self.PRIMITIVE_CASTERS[schema_type]
return PrimitiveCaster(schema, caster_callable)
elif schema.type in self.COMPLEX_CASTERS:
caster_class = self.COMPLEX_CASTERS[schema.type]
elif schema_type in self.COMPLEX_CASTERS:
caster_class = self.COMPLEX_CASTERS[schema_type]
return caster_class(schema, self)

6
openapi_core/deserializing/media_types/factories.py
View file

@ -20,11 +20,11 @@ class MediaTypeDeserializersFactory(object):
custom_deserializers = {}
self.custom_deserializers = custom_deserializers
def create(self, media_type):
def create(self, mimetype):
deserialize_callable = self.get_deserializer_callable(
media_type.mimetype)
mimetype)
return PrimitiveDeserializer(
media_type.mimetype, deserialize_callable)
mimetype, deserialize_callable)
def get_deserializer_callable(self, mimetype):
if mimetype in self.custom_deserializers:

20
openapi_core/deserializing/parameters/deserializers.py
View file

@ -2,7 +2,7 @@ from openapi_core.deserializing.exceptions import DeserializeError
from openapi_core.deserializing.parameters.exceptions import (
EmptyParameterValue,
)
from openapi_core.schema.parameters.enums import ParameterLocation
from openapi_core.schema.parameters import get_aslist, get_explode, get_style
class PrimitiveDeserializer(object):
@ -11,15 +11,19 @@ class PrimitiveDeserializer(object):
self.param = param
self.deserializer_callable = deserializer_callable
def __call__(self, value):
if (self.param.location == ParameterLocation.QUERY and value == "" and
not self.param.allow_empty_value):
raise EmptyParameterValue(
value, self.param.style, self.param.name)
self.aslist = get_aslist(self.param)
self.explode = get_explode(self.param)
self.style = get_style(self.param)
if not self.param.aslist or self.param.explode:
def __call__(self, value):
if (self.param['in'] == 'query' and value == "" and
not self.param.getkey('allowEmptyValue', False)):
raise EmptyParameterValue(
value, self.style, self.param['name'])
if not self.aslist or self.explode:
return value
try:
return self.deserializer_callable(value)
except (ValueError, TypeError, AttributeError):
raise DeserializeError(value, self.param.style)
raise DeserializeError(value, self.style)

18
openapi_core/deserializing/parameters/factories.py
View file

@ -3,24 +3,26 @@ import warnings
from openapi_core.deserializing.parameters.deserializers import (
PrimitiveDeserializer,
)
from openapi_core.schema.parameters.enums import ParameterStyle
from openapi_core.schema.parameters import get_style
class ParameterDeserializersFactory(object):
PARAMETER_STYLE_DESERIALIZERS = {
ParameterStyle.FORM: lambda x: x.split(','),
ParameterStyle.SIMPLE: lambda x: x.split(','),
ParameterStyle.SPACE_DELIMITED: lambda x: x.split(' '),
ParameterStyle.PIPE_DELIMITED: lambda x: x.split('|'),
'form': lambda x: x.split(','),
'simple': lambda x: x.split(','),
'spaceDelimited': lambda x: x.split(' '),
'pipeDelimited': lambda x: x.split('|'),
}
def create(self, param):
if param.deprecated:
if param.getkey('deprecated', False):
warnings.warn(
"{0} parameter is deprecated".format(param.name),
"{0} parameter is deprecated".format(param['name']),
DeprecationWarning,
)
deserialize_callable = self.PARAMETER_STYLE_DESERIALIZERS[param.style]
style = get_style(param)
deserialize_callable = self.PARAMETER_STYLE_DESERIALIZERS[style]
return PrimitiveDeserializer(param, deserialize_callable)

51
openapi_core/exceptions.py
View file

@ -1,5 +1,56 @@
"""OpenAPI core exceptions module"""
import attr
class OpenAPIError(Exception):
pass
class OpenAPIParameterError(OpenAPIError):
pass
class MissingParameterError(OpenAPIParameterError):
"""Missing parameter error"""
pass
@attr.s(hash=True)
class MissingParameter(MissingParameterError):
name = attr.ib()
def __str__(self):
return "Missing parameter (without default value): {0}".format(
self.name)
@attr.s(hash=True)
class MissingRequiredParameter(MissingParameterError):
name = attr.ib()
def __str__(self):
return "Missing required parameter: {0}".format(self.name)
class OpenAPIRequestBodyError(OpenAPIError):
pass
@attr.s(hash=True)
class MissingRequestBody(OpenAPIRequestBodyError):
request = attr.ib()
def __str__(self):
return "Missing required request body"
class OpenAPIResponseError(OpenAPIError):
pass
@attr.s(hash=True)
class MissingResponseContent(OpenAPIResponseError):
response = attr.ib()
def __str__(self):
return "Missing response content"

60
openapi_core/schema/components/factories.py
View file

@ -1,60 +0,0 @@
from openapi_core.compat import lru_cache
from openapi_core.schema.components.models import Components
from openapi_core.schema.extensions.generators import ExtensionsGenerator
from openapi_core.schema.schemas.generators import SchemasGenerator
from openapi_core.schema.security_schemes.generators import (
SecuritySchemesGenerator,
)
class ComponentsFactory(object):
def __init__(self, dereferencer, schemas_registry):
self.dereferencer = dereferencer
self.schemas_registry = schemas_registry
def create(self, components_spec):
components_deref = self.dereferencer.dereference(components_spec)
schemas_spec = components_deref.get('schemas', {})
responses_spec = components_deref.get('responses', {})
parameters_spec = components_deref.get('parameters', {})
request_bodies_spec = components_deref.get('requestBodies', {})
security_schemes_spec = components_deref.get('securitySchemes', {})
extensions = self.extensions_generator.generate(components_deref)
schemas = self.schemas_generator.generate(schemas_spec)
responses = self._generate_response(responses_spec)
parameters = self._generate_parameters(parameters_spec)
request_bodies = self._generate_request_bodies(request_bodies_spec)
security_schemes = self._generate_security_schemes(
security_schemes_spec)
return Components(
schemas=list(schemas), responses=responses, parameters=parameters,
request_bodies=request_bodies, security_schemes=security_schemes,
extensions=extensions,
)
@property
@lru_cache()
def schemas_generator(self):
return SchemasGenerator(self.dereferencer, self.schemas_registry)
@property
@lru_cache()
def extensions_generator(self):
return ExtensionsGenerator(self.dereferencer)
def _generate_response(self, responses_spec):
return responses_spec
def _generate_parameters(self, parameters_spec):
return parameters_spec
def _generate_request_bodies(self, request_bodies_spec):
return request_bodies_spec
def _generate_security_schemes(self, security_schemes_spec):
return SecuritySchemesGenerator(self.dereferencer).generate(
security_schemes_spec)

15
openapi_core/schema/components/models.py
View file

@ -1,15 +0,0 @@
class Components(object):
"""Represents an OpenAPI Components in a service."""
def __init__(
self, schemas=None, responses=None, parameters=None,
request_bodies=None, security_schemes=None, extensions=None):
self.schemas = schemas and dict(schemas) or {}
self.responses = responses and dict(responses) or {}
self.parameters = parameters and dict(parameters) or {}
self.request_bodies = request_bodies and dict(request_bodies) or {}
self.security_schemes = (
security_schemes and dict(security_schemes) or {}
)
self.extensions = extensions and dict(extensions) or {}

0
openapi_core/schema/contacts/__init__.py
View file

25
openapi_core/schema/contacts/factories.py
View file

@ -1,25 +0,0 @@
"""OpenAPI core contacts factories module"""
from openapi_core.compat import lru_cache
from openapi_core.schema.contacts.models import Contact
from openapi_core.schema.extensions.generators import ExtensionsGenerator
class ContactFactory(object):
def __init__(self, dereferencer):
self.dereferencer = dereferencer
def create(self, contact_spec):
contact_deref = self.dereferencer.dereference(contact_spec)
name = contact_deref.get('name')
url = contact_deref.get('url')
email = contact_deref.get('email')
extensions = self.extensions_generator.generate(contact_deref)
return Contact(name=name, url=url, email=email, extensions=extensions)
@property
@lru_cache()
def extensions_generator(self):
return ExtensionsGenerator(self.dereferencer)

11
openapi_core/schema/contacts/models.py
View file

@ -1,11 +0,0 @@
"""OpenAPI core contacts models module"""
class Contact(object):
def __init__(self, name=None, url=None, email=None, extensions=None):
self.name = name
self.url = url
self.email = email
self.extensions = extensions and dict(extensions) or {}

0
openapi_core/schema/content/__init__.py
View file

21
openapi_core/schema/content/factories.py
View file

@ -1,21 +0,0 @@
"""OpenAPI core content factories module"""
from openapi_core.compat import lru_cache
from openapi_core.schema.content.models import Content
from openapi_core.schema.media_types.generators import MediaTypeGenerator
class ContentFactory(object):
def __init__(self, dereferencer, schemas_registry):
self.dereferencer = dereferencer
self.schemas_registry = schemas_registry
def create(self, content_spec):
media_types = self.media_types_generator.generate(content_spec)
return Content(media_types)
@property
@lru_cache()
def media_types_generator(self):
return MediaTypeGenerator(self.dereferencer, self.schemas_registry)

5
openapi_core/schema/content/models.py
View file

@ -1,5 +0,0 @@
"""OpenAPI core content models module"""
class Content(dict):
pass

6
openapi_core/schema/exceptions.py
View file

@ -1,6 +0,0 @@
"""OpenAPI core schema exceptions module"""
from openapi_core.exceptions import OpenAPIError
class OpenAPIMappingError(OpenAPIError):
pass

0
openapi_core/schema/extensions/__init__.py
View file

16
openapi_core/schema/extensions/generators.py
View file

@ -1,16 +0,0 @@
"""OpenAPI core extensions generators module"""
from six import iteritems
from openapi_core.schema.extensions.models import Extension
class ExtensionsGenerator(object):
def __init__(self, dereferencer):
self.dereferencer = dereferencer
def generate(self, item_spec):
for field_name, value in iteritems(item_spec):
if not field_name.startswith('x-'):
continue
yield field_name, Extension(field_name, value)

9
openapi_core/schema/extensions/models.py
View file

@ -1,9 +0,0 @@
"""OpenAPI core extensions models module"""
class Extension(object):
"""Represents an OpenAPI Extension."""
def __init__(self, field_name, value=None):
self.field_name = field_name
self.value = value

0
openapi_core/schema/external_docs/__init__.py
View file

26
openapi_core/schema/external_docs/factories.py
View file

@ -1,26 +0,0 @@
"""OpenAPI core external docs factories module"""
from openapi_core.compat import lru_cache
from openapi_core.schema.extensions.generators import ExtensionsGenerator
from openapi_core.schema.external_docs.models import ExternalDocumentation
class ExternalDocumentationFactory(object):
def __init__(self, dereferencer):
self.dereferencer = dereferencer
def create(self, external_doc_spec):
url = external_doc_spec['url']
description = external_doc_spec.get('description')
extensions = self.extensions_generator.generate(external_doc_spec)
return ExternalDocumentation(
url,
description=description, extensions=extensions,
)
@property
@lru_cache()
def extensions_generator(self):
return ExtensionsGenerator(self.dereferencer)

11
openapi_core/schema/external_docs/models.py
View file

@ -1,11 +0,0 @@
"""OpenAPI core external docs models module"""
class ExternalDocumentation(object):
"""Represents an OpenAPI External Documentation."""
def __init__(self, url, description=None, extensions=None):
self.url = url
self.description = description
self.extensions = extensions and dict(extensions) or {}

0
openapi_core/schema/infos/__init__.py
View file

52
openapi_core/schema/infos/factories.py
View file

@ -1,52 +0,0 @@
"""OpenAPI core infos factories module"""
from openapi_core.compat import lru_cache
from openapi_core.schema.contacts.factories import ContactFactory
from openapi_core.schema.extensions.generators import ExtensionsGenerator
from openapi_core.schema.infos.models import Info
from openapi_core.schema.licenses.factories import LicenseFactory
class InfoFactory(object):
def __init__(self, dereferencer):
self.dereferencer = dereferencer
def create(self, info_spec):
info_deref = self.dereferencer.dereference(info_spec)
title = info_deref['title']
version = info_deref['version']
description = info_deref.get('description')
terms_of_service = info_deref.get('termsOfService')
extensions = self.extensions_generator.generate(info_deref)
contact = None
if 'contact' in info_deref:
contact_spec = info_deref.get('contact')
contact = self.contact_factory.create(contact_spec)
license = None
if 'license' in info_deref:
license_spec = info_deref.get('license')
license = self.license_factory.create(license_spec)
return Info(
title, version,
description=description, terms_of_service=terms_of_service,
contact=contact, license=license, extensions=extensions,
)
@property
@lru_cache()
def contact_factory(self):
return ContactFactory(self.dereferencer)
@property
@lru_cache()
def license_factory(self):
return LicenseFactory(self.dereferencer)
@property
@lru_cache()
def extensions_generator(self):
return ExtensionsGenerator(self.dereferencer)

17
openapi_core/schema/infos/models.py
View file

@ -1,17 +0,0 @@
"""OpenAPI core infos models module"""
class Info(object):
def __init__(
self, title, version, description=None, terms_of_service=None,
contact=None, license=None, extensions=None,
):
self.title = title
self.version = version
self.description = description
self.terms_of_service = terms_of_service
self.contact = contact
self.license = license
self.extensions = extensions and dict(extensions) or {}

0
openapi_core/schema/licenses/__init__.py
View file

24
openapi_core/schema/licenses/factories.py
View file

@ -1,24 +0,0 @@
"""OpenAPI core licenses factories module"""
from openapi_core.compat import lru_cache
from openapi_core.schema.extensions.generators import ExtensionsGenerator
from openapi_core.schema.licenses.models import License
class LicenseFactory(object):
def __init__(self, dereferencer):
self.dereferencer = dereferencer
def create(self, license_spec):
license_deref = self.dereferencer.dereference(license_spec)
name = license_deref['name']
url = license_deref.get('url')
extensions = self.extensions_generator.generate(license_deref)
return License(name, url=url, extensions=extensions)
@property
@lru_cache()
def extensions_generator(self):
return ExtensionsGenerator(self.dereferencer)

10
openapi_core/schema/licenses/models.py
View file

@ -1,10 +0,0 @@
"""OpenAPI core licenses models module"""
class License(object):
def __init__(self, name, url=None, extensions=None):
self.name = name
self.url = url
self.extensions = extensions and dict(extensions) or {}

0
openapi_core/schema/links/__init__.py
View file

44
openapi_core/schema/links/generators.py
View file

@ -1,44 +0,0 @@
"""OpenAPI core links generators module"""
from six import iteritems
from openapi_core.compat import lru_cache
from openapi_core.schema.links.models import Link
from openapi_core.schema.parameters.generators import ParametersGenerator
from openapi_core.schema.servers.generators import ServersGenerator
class LinksGenerator(object):
def __init__(self, dereferencer, schemas_registry):
self.dereferencer = dereferencer
self.schemas_registry = schemas_registry
def generate(self, links):
for link_name, link in iteritems(links):
link_deref = self.dereferencer.dereference(link)
operation_id = link_deref.get('operationId')
parameters = link_deref.get('parameters', {})
request_body = link_deref.get('requestBody') # string or dict
description = link_deref.get('description')
server_spec = link_deref.get('server')
server = self.servers_generator.generate(server_spec) \
if server_spec is not None \
else None
yield link_name, Link(
operation_id,
parameters,
request_body,
description,
server
)
@property
@lru_cache()
def parameters_generator(self):
return ParametersGenerator(self.dereferencer, self.schemas_registry)
@property
@lru_cache()
def servers_generator(self):
return ServersGenerator(self.dereferencer)

26
openapi_core/schema/links/models.py
View file

@ -1,26 +0,0 @@
"""OpenAPI core links models module"""
class Link(object):
"""Represents an OpenAPI Link."""
def __init__(
self,
operation_id,
parameters,
request_body,
description,
server
):
"""
request_body is assumed to be either a string (JSON, YAML or
runtime expression) or an object (deserialized JSON or YAML)
"""
self.operationId = operation_id
self.description = description
self.server = server
self.parameters = dict(parameters) if parameters else {}
self.request_body = request_body
def __getitem__(self, item):
return self.parameters[item]

0
openapi_core/schema/media_types/__init__.py
View file

16
openapi_core/schema/media_types/exceptions.py
View file

@ -1,16 +0,0 @@
import attr
from openapi_core.schema.exceptions import OpenAPIMappingError
class OpenAPIMediaTypeError(OpenAPIMappingError):
pass
@attr.s(hash=True)
class InvalidContentType(OpenAPIMediaTypeError):
mimetype = attr.ib()
def __str__(self):
return "Content for following mimetype not found: {0}".format(
self.mimetype)

40
openapi_core/schema/media_types/generators.py
View file

@ -1,40 +0,0 @@
"""OpenAPI core media types generators module"""
from six import iteritems
from openapi_core.compat import lru_cache
from openapi_core.schema.extensions.generators import ExtensionsGenerator
from openapi_core.schema.media_types.models import MediaType
class MediaTypeGenerator(object):
def __init__(self, dereferencer, schemas_registry):
self.dereferencer = dereferencer
self.schemas_registry = schemas_registry
def generate(self, content):
for mimetype, media_type in iteritems(content):
schema_spec = media_type.get('schema')
example_spec = media_type.get('example')
example_type = type(example_spec)
if example_type is dict:
example = self.dereferencer.dereference(example_spec)
else:
example = example_spec
extensions = self.extensions_generator.generate(media_type)
schema = None
if schema_spec:
schema, _ = self.schemas_registry.get_or_create(schema_spec)
yield mimetype, MediaType(
mimetype,
schema=schema, example=example, extensions=extensions,
)
@property
@lru_cache()
def extensions_generator(self):
return ExtensionsGenerator(self.dereferencer)

12
openapi_core/schema/media_types/models.py
View file

@ -1,12 +0,0 @@
"""OpenAPI core media types models module"""
class MediaType(object):
"""Represents an OpenAPI MediaType."""
def __init__(self, mimetype, schema=None, example=None, extensions=None):
self.mimetype = mimetype
self.schema = schema
self.example = example
self.extensions = extensions and dict(extensions) or {}

0
openapi_core/schema/operations/__init__.py
View file

17
openapi_core/schema/operations/exceptions.py
View file

@ -1,17 +0,0 @@
import attr
from openapi_core.schema.exceptions import OpenAPIMappingError
class OpenAPIOperationError(OpenAPIMappingError):
pass
@attr.s(hash=True)
class InvalidOperation(OpenAPIOperationError):
path_pattern = attr.ib()
http_method = attr.ib()
def __str__(self):
return "Unknown operation path {0} with method {1}".format(
self.path_pattern, self.http_method)

114
openapi_core/schema/operations/generators.py
View file

@ -1,114 +0,0 @@
# -*- coding: utf-8 -*-
"""OpenAPI core operations models module"""
from six import iteritems
from openapi_spec_validator.validators import PathItemValidator
from openapi_core.compat import lru_cache
from openapi_core.schema.extensions.generators import ExtensionsGenerator
from openapi_core.schema.external_docs.factories import (
ExternalDocumentationFactory,
)
from openapi_core.schema.operations.models import Operation
from openapi_core.schema.parameters.generators import ParametersGenerator
from openapi_core.schema.request_bodies.factories import RequestBodyFactory
from openapi_core.schema.responses.generators import ResponsesGenerator
from openapi_core.schema.security_requirements.generators import (
SecurityRequirementsGenerator,
)
from openapi_core.schema.servers.generators import ServersGenerator
class OperationsGenerator(object):
"""Represents an OpenAPI Operation in a service."""
def __init__(self, dereferencer, schemas_registry):
self.dereferencer = dereferencer
self.schemas_registry = schemas_registry
def generate(self, path_name, path):
path_deref = self.dereferencer.dereference(path)
for http_method, operation in iteritems(path_deref):
if http_method not in PathItemValidator.OPERATIONS:
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_from_list(
parameters_list)
operation_id = operation_deref.get('operationId')
tags_list = operation_deref.get('tags', [])
summary = operation_deref.get('summary')
description = operation_deref.get('description')
servers_spec = operation_deref.get('servers', [])
servers = self.servers_generator.generate(servers_spec)
security = None
if 'security' in operation_deref:
security_spec = operation_deref.get('security')
security = self.security_requirements_generator.generate(
security_spec)
extensions = self.extensions_generator.generate(operation_deref)
external_docs = None
if 'externalDocs' in operation_deref:
external_docs_spec = operation_deref.get('externalDocs')
external_docs = self.external_docs_factory.create(
external_docs_spec)
request_body = None
if 'requestBody' in operation_deref:
request_body_spec = operation_deref.get('requestBody')
request_body = self.request_body_factory.create(
request_body_spec)
yield (
http_method,
Operation(
http_method, path_name, responses, list(parameters),
summary=summary, description=description,
external_docs=external_docs, security=security,
request_body=request_body, deprecated=deprecated,
operation_id=operation_id, tags=list(tags_list),
servers=list(servers), extensions=extensions,
),
)
@property
@lru_cache()
def responses_generator(self):
return ResponsesGenerator(self.dereferencer, self.schemas_registry)
@property
@lru_cache()
def parameters_generator(self):
return ParametersGenerator(self.dereferencer, self.schemas_registry)
@property
@lru_cache()
def external_docs_factory(self):
return ExternalDocumentationFactory(self.dereferencer)
@property
@lru_cache()
def request_body_factory(self):
return RequestBodyFactory(self.dereferencer, self.schemas_registry)
@property
@lru_cache()
def security_requirements_generator(self):
return SecurityRequirementsGenerator(self.dereferencer)
@property
@lru_cache()
def servers_generator(self):
return ServersGenerator(self.dereferencer)
@property
@lru_cache()
def extensions_generator(self):
return ExtensionsGenerator(self.dereferencer)

30
openapi_core/schema/operations/models.py
View file

@ -1,30 +0,0 @@
# -*- coding: utf-8 -*-
"""OpenAPI core operations models module"""
class Operation(object):
"""Represents an OpenAPI Operation."""
def __init__(
self, http_method, path_name, responses, parameters,
summary=None, description=None, external_docs=None, security=None,
request_body=None, deprecated=False, operation_id=None, tags=None,
servers=None, extensions=None):
self.http_method = http_method
self.path_name = path_name
self.responses = dict(responses)
self.parameters = dict(parameters)
self.summary = summary
self.description = description
self.external_docs = external_docs
self.security = security and list(security)
self.request_body = request_body
self.deprecated = deprecated
self.operation_id = operation_id
self.tags = tags
self.servers = servers
self.extensions = extensions and dict(extensions) or {}
def __getitem__(self, name):
return self.parameters[name]

24
openapi_core/schema/parameters.py Normal file
View file

@ -0,0 +1,24 @@
def get_aslist(param):
return (
param.get('schema', None) and
param['schema']['type'] in ['array', 'object']
)
def get_style(param):
if 'style' in param:
return param['style']
# determine default
return (
'simple' if param['in'] in ['path', 'header'] else 'form'
)
def get_explode(param):
if 'explode' in param:
return param['explode']
# determine default
style = get_style(param)
return style == 'form'

0
openapi_core/schema/parameters/__init__.py
View file

25
openapi_core/schema/parameters/enums.py
View file

@ -1,25 +0,0 @@
"""OpenAPI core parameters enums module"""
from enum import Enum
class ParameterLocation(Enum):
PATH = 'path'
QUERY = 'query'
HEADER = 'header'
COOKIE = 'cookie'
@classmethod
def has_value(cls, value):
return (any(value == item.value for item in cls))
class ParameterStyle(Enum):
MATRIX = 'matrix'
LABEL = 'label'
FORM = 'form'
SIMPLE = 'simple'
SPACE_DELIMITED = 'spaceDelimited'
PIPE_DELIMITED = 'pipeDelimited'
DEEP_OBJECT = 'deepObject'

29
openapi_core/schema/parameters/exceptions.py
View file

@ -1,29 +0,0 @@
import attr
from openapi_core.schema.exceptions import OpenAPIMappingError
class OpenAPIParameterError(OpenAPIMappingError):
pass
class MissingParameterError(OpenAPIParameterError):
"""Missing parameter error"""
pass
@attr.s(hash=True)
class MissingParameter(MissingParameterError):
name = attr.ib()
def __str__(self):
return "Missing parameter (without default value): {0}".format(
self.name)
@attr.s(hash=True)
class MissingRequiredParameter(MissingParameterError):
name = attr.ib()
def __str__(self):
return "Missing required parameter: {0}".format(self.name)

54
openapi_core/schema/parameters/factories.py
View file

@ -1,54 +0,0 @@
"""OpenAPI core parameters factories module"""
from openapi_core.compat import lru_cache
from openapi_core.schema.content.factories import ContentFactory
from openapi_core.schema.extensions.generators import ExtensionsGenerator
from openapi_core.schema.parameters.models import Parameter
class ParameterFactory(object):
def __init__(self, dereferencer, schemas_registry):
self.dereferencer = dereferencer
self.schemas_registry = schemas_registry
def create(self, parameter_spec, parameter_name=None):
parameter_deref = self.dereferencer.dereference(parameter_spec)
parameter_name = parameter_name or parameter_deref['name']
parameter_in = parameter_deref.get('in', 'header')
allow_empty_value = parameter_deref.get('allowEmptyValue')
required = parameter_deref.get('required', False)
style = parameter_deref.get('style')
explode = parameter_deref.get('explode')
schema_spec = parameter_deref.get('schema', None)
schema = None
if schema_spec:
schema, _ = self.schemas_registry.get_or_create(schema_spec)
content_spec = parameter_deref.get('content', None)
content = None
if content_spec:
content = self.content_factory.create(content_spec)
extensions = self.extensions_generator.generate(parameter_deref)
return Parameter(
parameter_name, parameter_in,
schema=schema, required=required,
allow_empty_value=allow_empty_value,
style=style, explode=explode, content=content,
extensions=extensions,
)
@property
@lru_cache()
def content_factory(self):
return ContentFactory(self.dereferencer, self.schemas_registry)
@property
@lru_cache()
def extensions_generator(self):
return ExtensionsGenerator(self.dereferencer)

30
openapi_core/schema/parameters/generators.py
View file

@ -1,30 +0,0 @@
"""OpenAPI core parameters generators module"""
from six import iteritems
from openapi_core.compat import lru_cache
from openapi_core.schema.parameters.factories import ParameterFactory
class ParametersGenerator(object):
def __init__(self, dereferencer, schemas_registry):
self.dereferencer = dereferencer
self.schemas_registry = schemas_registry
def generate(self, parameters):
for parameter_name, parameter_spec in iteritems(parameters):
parameter = self.parameter_factory.create(
parameter_spec, parameter_name=parameter_name)
yield (parameter_name, parameter)
def generate_from_list(self, parameters_list):
for parameter_spec in parameters_list:
parameter = self.parameter_factory.create(parameter_spec)
yield (parameter.name, parameter)
@property
@lru_cache()
def parameter_factory(self):
return ParameterFactory(self.dereferencer, self.schemas_registry)

53
openapi_core/schema/parameters/models.py
View file

@ -1,53 +0,0 @@
"""OpenAPI core parameters models module"""
import logging
from openapi_core.schema.parameters.enums import (
ParameterLocation, ParameterStyle,
)
from openapi_core.schema.schemas.enums import SchemaType
log = logging.getLogger(__name__)
class Parameter(object):
"""Represents an OpenAPI operation Parameter."""
def __init__(
self, name, location, schema=None, required=False,
deprecated=False, allow_empty_value=False,
items=None, style=None, explode=None, content=None,
extensions=None):
self.name = name
self.location = ParameterLocation(location)
self.schema = schema
self.required = (
True if self.location == ParameterLocation.PATH else required
)
self.deprecated = deprecated
self.allow_empty_value = (
allow_empty_value if self.location == ParameterLocation.QUERY
else False
)
self.items = items
self.style = ParameterStyle(style or self.default_style)
self.explode = self.default_explode if explode is None else explode
self.content = content
self.extensions = extensions and dict(extensions) or {}
@property
def aslist(self):
return (
self.schema and
self.schema.type in [SchemaType.ARRAY, SchemaType.OBJECT]
)
@property
def default_style(self):
simple_locations = [ParameterLocation.PATH, ParameterLocation.HEADER]
return (
'simple' if self.location in simple_locations else "form"
)
@property
def default_explode(self):
return self.style == ParameterStyle.FORM

0
openapi_core/schema/paths/__init__.py
View file

15
openapi_core/schema/paths/exceptions.py
View file

@ -1,15 +0,0 @@
import attr
from openapi_core.schema.exceptions import OpenAPIMappingError
class OpenAPIPathError(OpenAPIMappingError):
pass
@attr.s(hash=True)
class InvalidPath(OpenAPIPathError):
path_pattern = attr.ib()
def __str__(self):
return "Unknown path {0}".format(self.path_pattern)

62
openapi_core/schema/paths/generators.py
View file

@ -1,62 +0,0 @@
"""OpenAPI core paths generators module"""
from six import iteritems
from openapi_core.compat import lru_cache
from openapi_core.schema.extensions.generators import ExtensionsGenerator
from openapi_core.schema.operations.generators import OperationsGenerator
from openapi_core.schema.parameters.generators import ParametersGenerator
from openapi_core.schema.paths.models import Path
from openapi_core.schema.servers.generators import ServersGenerator
class PathsGenerator(object):
def __init__(self, dereferencer, schemas_registry):
self.dereferencer = dereferencer
self.schemas_registry = schemas_registry
def generate(self, paths):
paths_deref = self.dereferencer.dereference(paths)
for path_name, path_spec in iteritems(paths_deref):
path_deref = self.dereferencer.dereference(path_spec)
parameters_list = path_deref.get('parameters', [])
summary = path_deref.get('summary')
description = path_deref.get('description')
servers_spec = path_deref.get('servers', [])
operations = self.operations_generator.generate(
path_name, path_deref)
servers = self.servers_generator.generate(servers_spec)
parameters = self.parameters_generator.generate_from_list(
parameters_list)
extensions = self.extensions_generator.generate(path_deref)
yield (
path_name,
Path(
path_name, list(operations), parameters=list(parameters),
summary=summary, description=description,
servers=list(servers), extensions=extensions,
),
)
@property
@lru_cache()
def operations_generator(self):
return OperationsGenerator(self.dereferencer, self.schemas_registry)
@property
@lru_cache()
def servers_generator(self):
return ServersGenerator(self.dereferencer)
@property
@lru_cache()
def parameters_generator(self):
return ParametersGenerator(self.dereferencer, self.schemas_registry)
@property
@lru_cache()
def extensions_generator(self):
return ExtensionsGenerator(self.dereferencer)

22
openapi_core/schema/paths/models.py
View file

@ -1,22 +0,0 @@
"""OpenAPI core paths models module"""
class Path(object):
"""Represents an OpenAPI Path."""
def __init__(
self, name, operations,
summary=None, description=None, parameters=None, servers=None,
extensions=None,
):
self.name = name
self.operations = dict(operations)
self.summary = summary
self.description = description
self.servers = servers
self.parameters = dict(parameters) if parameters else {}
self.extensions = extensions and dict(extensions) or {}
def __getitem__(self, http_method):
return self.operations[http_method]

0
openapi_core/schema/properties/__init__.py
View file

18
openapi_core/schema/properties/generators.py
View file

@ -1,18 +0,0 @@
"""OpenAPI core properties generators module"""
from six import iteritems
class PropertiesGenerator(object):
def __init__(self, dereferencer, schemas_registry):
self.dereferencer = dereferencer
self.schemas_registry = schemas_registry
def generate(self, properties):
for property_name, schema_spec in iteritems(properties):
schema = self._create_schema(schema_spec)
yield property_name, schema
def _create_schema(self, schema_spec):
schema, _ = self.schemas_registry.get_or_create(schema_spec)
return schema

0
openapi_core/schema/request_bodies/__init__.py
View file

15
openapi_core/schema/request_bodies/exceptions.py
View file

@ -1,15 +0,0 @@
import attr
from openapi_core.schema.exceptions import OpenAPIMappingError
class OpenAPIRequestBodyError(OpenAPIMappingError):
pass
@attr.s(hash=True)
class MissingRequestBody(OpenAPIRequestBodyError):
request = attr.ib()
def __str__(self):
return "Missing required request body"

36
openapi_core/schema/request_bodies/factories.py
View file

@ -1,36 +0,0 @@
"""OpenAPI core request bodies factories module"""
from openapi_core.compat import lru_cache
from openapi_core.schema.content.factories import ContentFactory
from openapi_core.schema.extensions.generators import ExtensionsGenerator
from openapi_core.schema.request_bodies.models import RequestBody
class RequestBodyFactory(object):
def __init__(self, dereferencer, schemas_registry):
self.dereferencer = dereferencer
self.schemas_registry = schemas_registry
def create(self, request_body_spec):
request_body_deref = self.dereferencer.dereference(
request_body_spec)
content_spec = request_body_deref['content']
content = self.content_factory.create(content_spec)
required = request_body_deref.get('required', False)
extensions = self.extensions_generator.generate(request_body_deref)
return RequestBody(
content,
required=required, extensions=extensions,
)
@property
@lru_cache()
def content_factory(self):
return ContentFactory(self.dereferencer, self.schemas_registry)
@property
@lru_cache()
def extensions_generator(self):
return ExtensionsGenerator(self.dereferencer)

11
openapi_core/schema/request_bodies/models.py
View file

@ -1,11 +0,0 @@
"""OpenAPI core request bodies models module"""
class RequestBody(object):
"""Represents an OpenAPI RequestBody."""
def __init__(self, content, required=False, extensions=None):
self.content = content
self.required = required
self.extensions = extensions and dict(extensions) or {}

0
openapi_core/schema/responses/__init__.py
View file

15
openapi_core/schema/responses/exceptions.py
View file

@ -1,15 +0,0 @@
import attr
from openapi_core.schema.exceptions import OpenAPIMappingError
class OpenAPIResponseError(OpenAPIMappingError):
pass
@attr.s(hash=True)
class MissingResponseContent(OpenAPIResponseError):
response = attr.ib()
def __str__(self):
return "Missing response content"

61
openapi_core/schema/responses/generators.py
View file

@ -1,61 +0,0 @@
"""OpenAPI core responses generators module"""
from six import iteritems
from openapi_core.compat import lru_cache
from openapi_core.schema.content.factories import ContentFactory
from openapi_core.schema.extensions.generators import ExtensionsGenerator
from openapi_core.schema.links.generators import LinksGenerator
from openapi_core.schema.parameters.generators import ParametersGenerator
from openapi_core.schema.responses.models import Response
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):
response_deref = self.dereferencer.dereference(response)
description = response_deref['description']
headers = response_deref.get('headers')
content_spec = response_deref.get('content')
links_dict = response_deref.get('links', {})
links = self.links_generator.generate(links_dict)
extensions = self.extensions_generator.generate(response_deref)
content = None
if content_spec:
content = self.content_factory.create(content_spec)
parameters = None
if headers:
parameters = self.parameters_generator.generate(headers)
yield http_status, Response(
http_status, description,
content=content, headers=parameters, links=links,
extensions=extensions,
)
@property
@lru_cache()
def content_factory(self):
return ContentFactory(self.dereferencer, self.schemas_registry)
@property
@lru_cache()
def parameters_generator(self):
return ParametersGenerator(self.dereferencer, self.schemas_registry)
@property
@lru_cache()
def links_generator(self):
return LinksGenerator(self.dereferencer, self.schemas_registry)
@property
@lru_cache()
def extensions_generator(self):
return ExtensionsGenerator(self.dereferencer)

15
openapi_core/schema/responses/models.py
View file

@ -1,15 +0,0 @@
"""OpenAPI core responses models module"""
class Response(object):
def __init__(
self, http_status, description, headers=None, content=None,
links=None, extensions=None):
self.http_status = http_status
self.description = description
self.headers = headers and dict(headers) or {}
self.content = content
self.links = links and dict(links) or {}
self.extensions = extensions and dict(extensions) or {}

22
openapi_core/schema/schemas.py Normal file
View file

@ -0,0 +1,22 @@
from __future__ import division
from six import iteritems
def get_all_properties(schema):
properties = schema.get('properties', {})
properties_dict = dict(iteritems(properties))
if 'allOf'not in schema:
return properties_dict
for subschema in schema / 'allOf':
subschema_props = get_all_properties(subschema)
properties_dict.update(subschema_props)
return properties_dict
def get_all_properties_names(schema):
all_properties = get_all_properties(schema)
return set(all_properties.keys())

0
openapi_core/schema/schemas/__init__.py
View file

28
openapi_core/schema/schemas/enums.py
View file

@ -1,28 +0,0 @@
"""OpenAPI core schemas enums module"""
from enum import Enum
class SchemaType(Enum):
ANY = None
INTEGER = 'integer'
NUMBER = 'number'
STRING = 'string'
BOOLEAN = 'boolean'
ARRAY = 'array'
OBJECT = 'object'
class SchemaFormat(Enum):
NONE = None
INT32 = 'int32'
INT64 = 'int64'
FLOAT = 'float'
DOUBLE = 'double'
BYTE = 'byte'
BINARY = 'binary'
DATE = 'date'
DATETIME = 'date-time'
PASSWORD = 'password'
UUID = 'uuid'

5
openapi_core/schema/schemas/exceptions.py
View file

@ -1,5 +0,0 @@
from openapi_core.schema.exceptions import OpenAPIMappingError
class OpenAPISchemaError(OpenAPIMappingError):
pass

175
openapi_core/schema/schemas/factories.py
View file

@ -1,175 +0,0 @@
"""OpenAPI core schemas factories module"""
import logging
from six import iteritems
from openapi_core.compat import lru_cache
from openapi_core.schema.extensions.generators import ExtensionsGenerator
from openapi_core.schema.properties.generators import PropertiesGenerator
from openapi_core.schema.schemas.models import Schema
from openapi_core.schema.schemas.types import Contribution, NoValue
log = logging.getLogger(__name__)
class SchemaFactory(object):
def __init__(self, dereferencer):
self.dereferencer = dereferencer
def create(self, schema_spec):
schema_deref = self.dereferencer.dereference(schema_spec)
schema_type = schema_deref.get('type', None)
schema_format = schema_deref.get('format')
required = schema_deref.get('required', False)
default = schema_deref.get('default', NoValue)
properties_spec = schema_deref.get('properties', None)
items_spec = schema_deref.get('items', None)
nullable = schema_deref.get('nullable', False)
enum = schema_deref.get('enum', None)
deprecated = schema_deref.get('deprecated', False)
all_of_spec = schema_deref.get('allOf', None)
one_of_spec = schema_deref.get('oneOf', None)
additional_properties_spec = schema_deref.get('additionalProperties',
True)
min_items = schema_deref.get('minItems', None)
max_items = schema_deref.get('maxItems', None)
min_length = schema_deref.get('minLength', None)
max_length = schema_deref.get('maxLength', None)
pattern = schema_deref.get('pattern', None)
unique_items = schema_deref.get('uniqueItems', None)
minimum = schema_deref.get('minimum', None)
maximum = schema_deref.get('maximum', None)
multiple_of = schema_deref.get('multipleOf', None)
exclusive_minimum = schema_deref.get('exclusiveMinimum', False)
exclusive_maximum = schema_deref.get('exclusiveMaximum', False)
min_properties = schema_deref.get('minProperties', None)
max_properties = schema_deref.get('maxProperties', None)
read_only = schema_deref.get('readOnly', False)
write_only = schema_deref.get('writeOnly', False)
extensions = self.extensions_generator.generate(schema_deref)
properties = None
if properties_spec:
properties = self.properties_generator.generate(properties_spec)
all_of = []
if all_of_spec:
all_of = list(map(self.create, all_of_spec))
one_of = []
if one_of_spec:
one_of = list(map(self.create, one_of_spec))
items = None
if items_spec:
items = self._create_items(items_spec)
additional_properties = additional_properties_spec
if isinstance(additional_properties_spec, dict):
additional_properties = self.create(additional_properties_spec)
return Schema(
schema_type=schema_type, properties=properties,
items=items, schema_format=schema_format, required=required,
default=default, nullable=nullable, enum=enum,
deprecated=deprecated, all_of=all_of, one_of=one_of,
additional_properties=additional_properties,
min_items=min_items, max_items=max_items, min_length=min_length,
max_length=max_length, pattern=pattern, unique_items=unique_items,
minimum=minimum, maximum=maximum, multiple_of=multiple_of,
exclusive_maximum=exclusive_maximum,
exclusive_minimum=exclusive_minimum,
min_properties=min_properties, max_properties=max_properties,
read_only=read_only, write_only=write_only, extensions=extensions,
_source=schema_deref,
)
@property
@lru_cache()
def extensions_generator(self):
return ExtensionsGenerator(self.dereferencer)
@property
@lru_cache()
def properties_generator(self):
return PropertiesGenerator(self.dereferencer, self)
def _create_items(self, items_spec):
return self.create(items_spec)
class SchemaDictFactory(object):
contributions = (
Contribution('type', src_prop_attr='value'),
Contribution('format'),
Contribution('properties', is_dict=True, dest_default={}),
Contribution('required', dest_default=[]),
Contribution('default'),
Contribution('nullable', dest_default=False),
Contribution(
'all_of',
dest_prop_name='allOf', is_list=True, dest_default=[],
),
Contribution(
'one_of',
dest_prop_name='oneOf', is_list=True, dest_default=[],
),
Contribution(
'additional_properties',
dest_prop_name='additionalProperties', dest_default=True,
),
Contribution('min_items', dest_prop_name='minItems'),
Contribution('max_items', dest_prop_name='maxItems'),
Contribution('min_length', dest_prop_name='minLength'),
Contribution('max_length', dest_prop_name='maxLength'),
Contribution('pattern', src_prop_attr='pattern'),
Contribution(
'unique_items',
dest_prop_name='uniqueItems', dest_default=False,
),
Contribution('minimum'),
Contribution('maximum'),
Contribution('multiple_of', dest_prop_name='multipleOf'),
Contribution(
'exclusive_minimum',
dest_prop_name='exclusiveMinimum', dest_default=False,
),
Contribution(
'exclusive_maximum',
dest_prop_name='exclusiveMaximum', dest_default=False,
),
Contribution('min_properties', dest_prop_name='minProperties'),
Contribution('max_properties', dest_prop_name='maxProperties'),
)
def create(self, schema):
schema_dict = {}
for contrib in self.contributions:
self._contribute(schema, schema_dict, contrib)
return schema_dict
def _contribute(self, schema, schema_dict, contrib):
def src_map(x):
return getattr(x, '__dict__')
src_val = getattr(schema, contrib.src_prop_name)
if src_val and contrib.src_prop_attr:
src_val = getattr(src_val, contrib.src_prop_attr)
if contrib.is_list:
src_val = list(map(src_map, src_val))
if contrib.is_dict:
src_val = dict(
(k, src_map(v))
for k, v in iteritems(src_val)
)
if src_val == contrib.dest_default:
return
dest_prop_name = contrib.dest_prop_name or contrib.src_prop_name
schema_dict[dest_prop_name] = src_val

20
openapi_core/schema/schemas/generators.py
View file

@ -1,20 +0,0 @@
"""OpenAPI core schemas generators module"""
import logging
from six import iteritems
log = logging.getLogger(__name__)
class SchemasGenerator(object):
def __init__(self, dereferencer, schemas_registry):
self.dereferencer = dereferencer
self.schemas_registry = schemas_registry
def generate(self, schemas_spec):
schemas_deref = self.dereferencer.dereference(schemas_spec)
for schema_name, schema_spec in iteritems(schemas_deref):
schema, _ = self.schemas_registry.get_or_create(schema_spec)
yield schema_name, schema

95
openapi_core/schema/schemas/models.py
View file

@ -1,95 +0,0 @@
"""OpenAPI core schemas models module"""
import attr
import logging
import re
from openapi_core.schema.schemas.enums import SchemaType
from openapi_core.schema.schemas.types import NoValue
log = logging.getLogger(__name__)
@attr.s
class Format(object):
unmarshal = attr.ib()
validate = attr.ib()
class Schema(object):
"""Represents an OpenAPI Schema."""
def __init__(
self, schema_type=None, properties=None, items=None,
schema_format=None, required=None, default=NoValue, nullable=False,
enum=None, deprecated=False, all_of=None, one_of=None,
additional_properties=True, min_items=None, max_items=None,
min_length=None, max_length=None, pattern=None, unique_items=False,
minimum=None, maximum=None, multiple_of=None,
exclusive_minimum=False, exclusive_maximum=False,
min_properties=None, max_properties=None,
read_only=False, write_only=False, extensions=None,
_source=None):
self.type = SchemaType(schema_type)
self.properties = properties and dict(properties) or {}
self.items = items
self.format = schema_format
self.required = required or []
self.default = default
self.nullable = nullable
self.enum = enum
self.deprecated = deprecated
self.all_of = all_of and list(all_of) or []
self.one_of = one_of and list(one_of) or []
self.additional_properties = additional_properties
self.min_items = int(min_items) if min_items is not None else None
self.max_items = int(max_items) if max_items is not None else None
self.min_length = int(min_length) if min_length is not None else None
self.max_length = int(max_length) if max_length is not None else None
self.pattern = pattern and re.compile(pattern) or None
self.unique_items = unique_items
self.minimum = int(minimum) if minimum is not None else None
self.maximum = int(maximum) if maximum is not None else None
self.multiple_of = int(multiple_of)\
if multiple_of is not None else None
self.exclusive_minimum = exclusive_minimum
self.exclusive_maximum = exclusive_maximum
self.min_properties = int(min_properties)\
if min_properties is not None else None
self.max_properties = int(max_properties)\
if max_properties is not None else None
self.read_only = read_only
self.write_only = write_only
self.extensions = extensions and dict(extensions) or {}
self._all_required_properties_cache = None
self._all_optional_properties_cache = None
self._source = _source
@property
def __dict__(self):
return self._source or self.to_dict()
def to_dict(self):
from openapi_core.schema.schemas.factories import SchemaDictFactory
return SchemaDictFactory().create(self)
def __getitem__(self, name):
return self.properties[name]
def has_default(self):
return self.default is not NoValue
def get_all_properties(self):
properties = self.properties.copy()
for subschema in self.all_of:
subschema_props = subschema.get_all_properties()
properties.update(subschema_props)
return properties
def get_all_properties_names(self):
all_properties = self.get_all_properties()
return set(all_properties.keys())

32
openapi_core/schema/schemas/registries.py
View file

@ -1,32 +0,0 @@
"""OpenAPI core schemas registries module"""
import logging
from lazy_object_proxy import Proxy
from openapi_core.schema.schemas.factories import SchemaFactory
from openapi_core.schema.schemas.util import dicthash
log = logging.getLogger(__name__)
class SchemaRegistry(SchemaFactory):
def __init__(self, dereferencer):
super(SchemaRegistry, self).__init__(dereferencer)
self._schemas = {}
def get_or_create(self, schema_spec):
schema_hash = dicthash(schema_spec)
schema_deref = self.dereferencer.dereference(schema_spec)
if schema_hash in self._schemas:
return self._schemas[schema_hash], False
if '$ref' in schema_spec:
schema = Proxy(lambda: self.create(schema_deref))
else:
schema = self.create(schema_deref)
self._schemas[schema_hash] = schema
return schema, True

14
openapi_core/schema/schemas/types.py
View file

@ -1,14 +0,0 @@
import attr
NoValue = object()
@attr.s(hash=True)
class Contribution(object):
src_prop_name = attr.ib()
src_prop_attr = attr.ib(default=None)
dest_prop_name = attr.ib(default=None)
is_list = attr.ib(default=False)
is_dict = attr.ib(default=False)
dest_default = attr.ib(default=None)

15
openapi_core/schema/schemas/util.py
View file

@ -1,15 +0,0 @@
"""OpenAPI core schemas util module"""
from distutils.util import strtobool
from six import string_types
from json import dumps
def forcebool(val):
if isinstance(val, string_types):
val = strtobool(val)
return bool(val)
def dicthash(d):
return hash(dumps(d, sort_keys=True))

0
openapi_core/schema/security_requirements/__init__.py
View file

15
openapi_core/schema/security_requirements/generators.py
View file

@ -1,15 +0,0 @@
"""OpenAPI core security requirements generators module"""
from openapi_core.schema.security_requirements.models import (
SecurityRequirement,
)
class SecurityRequirementsGenerator(object):
def __init__(self, dereferencer):
self.dereferencer = dereferencer
def generate(self, security_spec):
security_deref = self.dereferencer.dereference(security_spec)
for security_requirement_spec in security_deref:
yield SecurityRequirement(security_requirement_spec)

6
openapi_core/schema/security_requirements/models.py
View file

@ -1,6 +0,0 @@
"""OpenAPI core security requirements models module"""
class SecurityRequirement(dict):
"""Represents an OpenAPI Security Requirement."""
pass

0
openapi_core/schema/security_schemes/__init__.py
View file

27
openapi_core/schema/security_schemes/enums.py
View file

@ -1,27 +0,0 @@
"""OpenAPI core security schemes enums module"""
from enum import Enum
class SecuritySchemeType(Enum):
API_KEY = 'apiKey'
HTTP = 'http'
OAUTH2 = 'oauth2'
OPEN_ID_CONNECT = 'openIdConnect'
class ApiKeyLocation(Enum):
QUERY = 'query'
HEADER = 'header'
COOKIE = 'cookie'
@classmethod
def has_value(cls, value):
return (any(value == item.value for item in cls))
class HttpAuthScheme(Enum):
BASIC = 'basic'
BEARER = 'bearer'

37
openapi_core/schema/security_schemes/generators.py
View file

@ -1,37 +0,0 @@
"""OpenAPI core security schemes generators module"""
import logging
from six import iteritems
from openapi_core.schema.security_schemes.models import SecurityScheme
log = logging.getLogger(__name__)
class SecuritySchemesGenerator(object):
def __init__(self, dereferencer):
self.dereferencer = dereferencer
def generate(self, security_schemes_spec):
security_schemes_deref = self.dereferencer.dereference(
security_schemes_spec)
for scheme_name, scheme_spec in iteritems(security_schemes_deref):
scheme_deref = self.dereferencer.dereference(scheme_spec)
scheme_type = scheme_deref['type']
description = scheme_deref.get('description')
name = scheme_deref.get('name')
apikey_in = scheme_deref.get('in')
scheme = scheme_deref.get('scheme')
bearer_format = scheme_deref.get('bearerFormat')
flows = scheme_deref.get('flows')
open_id_connect_url = scheme_deref.get('openIdConnectUrl')
scheme = SecurityScheme(
scheme_type, description=description, name=name,
apikey_in=apikey_in, scheme=scheme,
bearer_format=bearer_format, flows=flows,
open_id_connect_url=open_id_connect_url,
)
yield scheme_name, scheme

22
openapi_core/schema/security_schemes/models.py
View file

@ -1,22 +0,0 @@
"""OpenAPI core security schemes models module"""
from openapi_core.schema.security_schemes.enums import (
SecuritySchemeType, ApiKeyLocation, HttpAuthScheme,
)
class SecurityScheme(object):
"""Represents an OpenAPI Security Scheme."""
def __init__(
self, scheme_type, description=None, name=None, apikey_in=None,
scheme=None, bearer_format=None, flows=None,
open_id_connect_url=None,
):
self.type = SecuritySchemeType(scheme_type)
self.description = description
self.name = name
self.apikey_in = apikey_in and ApiKeyLocation(apikey_in)
self.scheme = scheme and HttpAuthScheme(scheme)
self.bearer_format = bearer_format
self.flows = flows
self.open_id_connect_url = open_id_connect_url

24
openapi_core/schema/servers.py Normal file
View file

@ -0,0 +1,24 @@
from __future__ import division
from six import iteritems
def is_absolute(url):
return url.startswith('//') or '://' in url
def get_server_default_variables(server):
if 'variables' not in server:
return {}
defaults = {}
variables = server / 'variables'
for name, variable in iteritems(variables):
defaults[name] = variable['default']
return defaults
def get_server_url(server, **variables):
if not variables:
variables = get_server_default_variables(server)
return server['url'].format(**variables)

0
openapi_core/schema/servers/__init__.py
View file

16
openapi_core/schema/servers/exceptions.py
View file

@ -1,16 +0,0 @@
import attr
from openapi_core.schema.exceptions import OpenAPIMappingError
class OpenAPIServerError(OpenAPIMappingError):
pass
@attr.s(hash=True)
class InvalidServer(OpenAPIServerError):
full_url_pattern = attr.ib()
def __str__(self):
return "Invalid request server {0}".format(
self.full_url_pattern)

67
openapi_core/schema/servers/generators.py
View file

@ -1,67 +0,0 @@
"""OpenAPI core servers generators module"""
from six import iteritems
from openapi_core.compat import lru_cache
from openapi_core.schema.extensions.generators import ExtensionsGenerator
from openapi_core.schema.servers.models import Server, ServerVariable
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', {})
description = server_spec.get('description')
extensions = self.extensions_generator.generate(server_spec)
variables = None
if variables_spec:
variables = self.variables_generator.generate(variables_spec)
yield Server(
url,
variables=variables, description=description,
extensions=extensions,
)
@property
@lru_cache()
def variables_generator(self):
return ServerVariablesGenerator(self.dereferencer)
@property
@lru_cache()
def extensions_generator(self):
return ExtensionsGenerator(self.dereferencer)
class ServerVariablesGenerator(object):
def __init__(self, dereferencer):
self.dereferencer = dereferencer
def generate(self, variables_spec):
variables_deref = self.dereferencer.dereference(variables_spec)
for variable_name, variable_spec in iteritems(variables_deref):
default = variable_spec['default']
enum = variable_spec.get('enum')
extensions = self.extensions_generator.generate(variable_spec)
variable = ServerVariable(
variable_name, default,
enum=enum, extensions=extensions,
)
yield variable_name, variable
@property
@lru_cache()
def extensions_generator(self):
return ExtensionsGenerator(self.dereferencer)

43
openapi_core/schema/servers/models.py
View file

@ -1,43 +0,0 @@
"""OpenAPI core servers models module"""
from six import iteritems
class Server(object):
def __init__(self, url, variables=None, description=None, extensions=None):
self.url = url
self.variables = variables and dict(variables) or {}
self.description = description
self.extensions = extensions and dict(extensions) 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)
def is_absolute(self, url=None):
if url is None:
url = self.url
return url.startswith('//') or '://' in url
class ServerVariable(object):
def __init__(self, name, default, enum=None, extensions=None):
self.name = name
self.default = default
self.enum = enum and list(enum) or []
self.extensions = extensions and dict(extensions) or {}

8
openapi_core/schema/specs.py Normal file
View file

@ -0,0 +1,8 @@
from __future__ import division
from openapi_core.schema.servers import get_server_url
def get_spec_url(spec, index=0):
servers = spec / 'servers'
return get_server_url(servers / 0)

0
openapi_core/schema/specs/__init__.py
View file

93
openapi_core/schema/specs/factories.py
View file

@ -1,93 +0,0 @@
# -*- coding: utf-8 -*-
"""OpenAPI core specs factories module"""
from openapi_spec_validator.validators import Dereferencer
from openapi_core.compat import lru_cache
from openapi_core.schema.components.factories import ComponentsFactory
from openapi_core.schema.extensions.generators import ExtensionsGenerator
from openapi_core.schema.infos.factories import InfoFactory
from openapi_core.schema.paths.generators import PathsGenerator
from openapi_core.schema.schemas.registries import SchemaRegistry
from openapi_core.schema.security_requirements.generators import (
SecurityRequirementsGenerator,
)
from openapi_core.schema.servers.generators import ServersGenerator
from openapi_core.schema.specs.models import Spec
class SpecFactory(object):
def __init__(self, spec_resolver):
self.spec_resolver = spec_resolver
def create(self, spec_dict, spec_url=''):
spec_dict_deref = self.dereferencer.dereference(spec_dict)
info_spec = spec_dict_deref.get('info', {})
servers_spec = spec_dict_deref.get('servers', [])
paths = spec_dict_deref.get('paths', {})
components_spec = spec_dict_deref.get('components', {})
security_spec = spec_dict_deref.get('security', [])
if not servers_spec:
servers_spec = [
{'url': '/'},
]
extensions = self.extensions_generator.generate(spec_dict_deref)
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)
security = self.security_requirements_generator.generate(
security_spec)
spec = Spec(
info, list(paths), servers=list(servers), components=components,
security=list(security), extensions=extensions,
_resolver=self.spec_resolver,
)
return spec
@property
@lru_cache()
def dereferencer(self):
return Dereferencer(self.spec_resolver)
@property
@lru_cache()
def schemas_registry(self):
return SchemaRegistry(self.dereferencer)
@property
@lru_cache()
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):
return PathsGenerator(self.dereferencer, self.schemas_registry)
@property
@lru_cache()
def components_factory(self):
return ComponentsFactory(self.dereferencer, self.schemas_registry)
@property
@lru_cache()
def security_requirements_generator(self):
return SecurityRequirementsGenerator(self.dereferencer)
@property
@lru_cache()
def extensions_generator(self):
return ExtensionsGenerator(self.dereferencer)

70
openapi_core/schema/specs/models.py
View file

@ -1,70 +0,0 @@
# -*- coding: utf-8 -*-
"""OpenAPI core specs models module"""
import logging
from openapi_core.compat import partialmethod
from openapi_core.schema.operations.exceptions import InvalidOperation
from openapi_core.schema.paths.exceptions import InvalidPath
from openapi_core.schema.servers.exceptions import InvalidServer
log = logging.getLogger(__name__)
class Spec(object):
"""Represents an OpenAPI Specification for a service."""
def __init__(
self, info, paths, servers=None, components=None,
security=None, extensions=None, _resolver=None):
self.info = info
self.paths = paths and dict(paths)
self.servers = servers or []
self.components = components
self.security = security
self.extensions = extensions and dict(extensions) or {}
self._resolver = _resolver
def __getitem__(self, path_pattern):
return self.get_path(path_pattern)
@property
def default_url(self):
return self.servers[0].default_url
def get_server(self, full_url_pattern):
for spec_server in self.servers:
if spec_server.default_url in full_url_pattern:
return spec_server
raise InvalidServer(full_url_pattern)
def get_server_url(self, index=0):
return self.servers[index].default_url
def get_path(self, path_pattern):
try:
return self.paths[path_pattern]
except KeyError:
raise InvalidPath(path_pattern)
def get_operation(self, path_pattern, http_method):
try:
return self.paths[path_pattern].operations[http_method]
except KeyError:
raise InvalidOperation(path_pattern, http_method)
def get_schema(self, name):
return self.components.schemas[name]
# operations shortcuts
get = partialmethod(get_operation, http_method='get')
put = partialmethod(get_operation, http_method='put')
post = partialmethod(get_operation, http_method='post')
delete = partialmethod(get_operation, http_method='delete')
options = partialmethod(get_operation, http_method='options')
head = partialmethod(get_operation, http_method='head')
patch = partialmethod(get_operation, http_method='patch')

10
openapi_core/security/factories.py
View file

@ -1,4 +1,3 @@
from openapi_core.schema.security_schemes.enums import SecuritySchemeType
from openapi_core.security.providers import (
ApiKeyProvider, HttpProvider, UnsupportedProvider,
)
@ -7,13 +6,14 @@ from openapi_core.security.providers import (
class SecurityProviderFactory(object):
PROVIDERS = {
SecuritySchemeType.API_KEY: ApiKeyProvider,
SecuritySchemeType.HTTP: HttpProvider,
'apiKey': ApiKeyProvider,
'http': HttpProvider,
}
def create(self, scheme):
if scheme.type == SecuritySchemeType.API_KEY:
scheme_type = scheme['type']
if scheme_type == 'apiKey':
return ApiKeyProvider(scheme)
elif scheme.type == SecuritySchemeType.HTTP:
elif scheme_type == 'http':
return HttpProvider(scheme)
return UnsupportedProvider(scheme)

11
openapi_core/security/providers.py
View file

@ -18,10 +18,12 @@ class UnsupportedProvider(BaseProvider):
class ApiKeyProvider(BaseProvider):
def __call__(self, request):
source = getattr(request.parameters, self.scheme.apikey_in.value)
if self.scheme.name not in source:
name = self.scheme['name']
location = self.scheme['in']
source = getattr(request.parameters, location)
if name not in source:
raise SecurityError("Missing api key parameter.")
return source.get(self.scheme.name)
return source[name]
class HttpProvider(BaseProvider):
@ -35,7 +37,8 @@ class HttpProvider(BaseProvider):
except ValueError:
raise SecurityError('Could not parse authorization header.')
if auth_type.lower() != self.scheme.scheme.value:
scheme = self.scheme['scheme']
if auth_type.lower() != scheme:
raise SecurityError(
'Unknown authorization method %s' % auth_type)

2
openapi_core/shortcuts.py
View file

@ -1,6 +1,6 @@
"""OpenAPI core shortcuts module"""
# backward compatibility
from openapi_core.schema.shortcuts import create_spec
from openapi_core.spec.shortcuts import create_spec
from openapi_core.validation.request.shortcuts import (
spec_validate_body as validate_body,
spec_validate_parameters as validate_parameters,

0
openapi_core/schema/components/__init__.py → openapi_core/spec/__init__.py
View file

23
openapi_core/spec/accessors.py Normal file
View file

@ -0,0 +1,23 @@
from contextlib import contextmanager
from dictpath.accessors import DictOrListAccessor
class SpecAccessor(DictOrListAccessor):
def __init__(self, dict_or_list, dereferencer):
super(SpecAccessor, self).__init__(dict_or_list)
self.dereferencer = dereferencer
@contextmanager
def open(self, parts):
content = self.dict_or_list
for part in parts:
content = content[part]
if '$ref' in content:
content = self.dereferencer.dereference(
content)
try:
yield content
finally:
pass

14
openapi_core/spec/paths.py Normal file
View file

@ -0,0 +1,14 @@
from dictpath.paths import AccessorPath
from openapi_core.spec.accessors import SpecAccessor
SPEC_SEPARATOR = '#'
class SpecPath(AccessorPath):
@classmethod
def from_spec(cls, spec_dict, dereferencer=None, *args, **kwargs):
separator = kwargs.pop('separator', SPEC_SEPARATOR)
accessor = SpecAccessor(spec_dict, dereferencer)
return cls(accessor, *args, separator=separator)

9
openapi_core/schema/shortcuts.py → openapi_core/spec/shortcuts.py
View file

@ -1,10 +1,11 @@
"""OpenAPI core schema shortcuts module"""
"""OpenAPI core spec shortcuts module"""
from jsonschema.validators import RefResolver
from openapi_spec_validator import (
default_handlers, openapi_v3_spec_validator,
)
from openapi_spec_validator.validators import Dereferencer
from openapi_core.schema.specs.factories import SpecFactory
from openapi_core.spec.paths import SpecPath
def create_spec(
@ -16,5 +17,5 @@ def create_spec(
spec_resolver = RefResolver(
spec_url, spec_dict, handlers=handlers)
spec_factory = SpecFactory(spec_resolver)
return spec_factory.create(spec_dict, spec_url=spec_url)
dereferencer = Dereferencer(spec_resolver)
return SpecPath.from_spec(spec_dict, dereferencer)

13
openapi_core/templating/media_types/finders.py
View file

@ -1,8 +1,7 @@
"""OpenAPI core templating media types finders module"""
from __future__ import division
import fnmatch
from six import iteritems
from openapi_core.templating.media_types.exceptions import MediaTypeNotFound
@ -12,13 +11,11 @@ class MediaTypeFinder(object):
self.content = content
def find(self, request):
try:
return self.content[request.mimetype]
except KeyError:
pass
if request.mimetype in self.content:
return self.content / request.mimetype, request.mimetype
for key, value in iteritems(self.content):
for key, value in self.content.items():
if fnmatch.fnmatch(request.mimetype, key):
return value
return value, key
raise MediaTypeNotFound(request.mimetype, list(self.content.keys()))

27
openapi_core/templating/paths/finders.py
View file

@ -1,8 +1,10 @@
"""OpenAPI core templating paths finders module"""
from __future__ import division
from more_itertools import peekable
from six import iteritems
from six.moves.urllib.parse import urljoin, urlparse
from openapi_core.schema.servers import is_absolute
from openapi_core.templating.datatypes import TemplateResult
from openapi_core.templating.util import parse, search
from openapi_core.templating.paths.exceptions import (
@ -40,7 +42,8 @@ class PathFinder(object):
def _get_paths_iter(self, full_url_pattern):
template_paths = []
for path_pattern, path in iteritems(self.spec.paths):
paths = self.spec / 'paths'
for path_pattern, path in paths.items():
# simple path.
# Return right away since it is always the most concrete
if full_url_pattern.endswith(path_pattern):
@ -59,22 +62,24 @@ class PathFinder(object):
def _get_operations_iter(self, request_method, paths_iter):
for path, path_result in paths_iter:
if request_method not in path.operations:
if request_method not in path:
continue
operation = path.operations[request_method]
operation = path / request_method
yield (path, operation, path_result)
def _get_servers_iter(self, full_url_pattern, ooperations_iter):
for path, operation, path_result in ooperations_iter:
servers = path.servers or operation.servers or self.spec.servers
servers = path.get('servers', None) or \
operation.get('servers', None) or \
self.spec.get('servers', [{'url': '/'}])
for server in servers:
server_url_pattern = full_url_pattern.rsplit(
path_result.resolved, 1)[0]
server_url = server.url
if not server.is_absolute():
server_url = server['url']
if not is_absolute(server_url):
# relative to absolute url
if self.base_url is not None:
server_url = urljoin(self.base_url, server.url)
server_url = urljoin(self.base_url, server['url'])
# if no base url check only path part
else:
server_url_pattern = urlparse(server_url_pattern).path
@ -82,17 +87,17 @@ class PathFinder(object):
server_url = server_url[:-1]
# simple path
if server_url_pattern == server_url:
server_result = TemplateResult(server.url, {})
server_result = TemplateResult(server['url'], {})
yield (
path, operation, server,
path_result, server_result,
)
# template path
else:
result = parse(server.url, server_url_pattern)
result = parse(server['url'], server_url_pattern)
if result:
server_result = TemplateResult(
server.url, result.named)
server['url'], result.named)
yield (
path, operation, server,
path_result, server_result,

12
openapi_core/templating/responses/finders.py
View file

@ -1,3 +1,5 @@
from __future__ import division
from openapi_core.templating.responses.exceptions import ResponseNotFound
@ -7,17 +9,15 @@ class ResponseFinder(object):
self.responses = responses
def find(self, http_status='default'):
try:
return self.responses[http_status]
except KeyError:
pass
if http_status in self.responses:
return self.responses / http_status
# try range
http_status_range = '{0}XX'.format(http_status[0])
if http_status_range in self.responses:
return self.responses[http_status_range]
return self.responses / http_status_range
if 'default' not in self.responses:
raise ResponseNotFound(http_status, self.responses)
return self.responses['default']
return self.responses / 'default'

1
openapi_core/types.py Normal file
View file

@ -0,0 +1 @@
NoValue = object()

43
openapi_core/unmarshalling/schemas/factories.py
View file

@ -2,8 +2,6 @@ import warnings
from openapi_schema_validator import OAS30Validator
from openapi_core.schema.schemas.enums import SchemaType, SchemaFormat
from openapi_core.schema.schemas.models import Schema
from openapi_core.unmarshalling.schemas.enums import UnmarshalContext
from openapi_core.unmarshalling.schemas.exceptions import (
FormatterNotFoundError,
@ -18,15 +16,15 @@ from openapi_core.unmarshalling.schemas.unmarshallers import (
class SchemaUnmarshallersFactory(object):
PRIMITIVE_UNMARSHALLERS = {
SchemaType.STRING: StringUnmarshaller,
SchemaType.INTEGER: IntegerUnmarshaller,
SchemaType.NUMBER: NumberUnmarshaller,
SchemaType.BOOLEAN: BooleanUnmarshaller,
'string': StringUnmarshaller,
'integer': IntegerUnmarshaller,
'number': NumberUnmarshaller,
'boolean': BooleanUnmarshaller,
}
COMPLEX_UNMARSHALLERS = {
SchemaType.ARRAY: ArrayUnmarshaller,
SchemaType.OBJECT: ObjectUnmarshaller,
SchemaType.ANY: AnyUnmarshaller,
'array': ArrayUnmarshaller,
'object': ObjectUnmarshaller,
'any': AnyUnmarshaller,
}
CONTEXT_VALIDATION = {
@ -46,12 +44,13 @@ class SchemaUnmarshallersFactory(object):
def create(self, schema, type_override=None):
"""Create unmarshaller from the schema."""
if not isinstance(schema, Schema):
raise TypeError("schema not type of Schema")
if schema.deprecated:
if schema is None:
raise TypeError("Invalid schema")
if schema.getkey('deprecated', False):
warnings.warn("The schema is deprecated", DeprecationWarning)
schema_type = type_override or schema.type
schema_type = type_override or schema.getkey('type', 'any')
if schema_type in self.PRIMITIVE_UNMARSHALLERS:
klass = self.PRIMITIVE_UNMARSHALLERS[schema_type]
kwargs = dict(schema=schema)
@ -63,22 +62,21 @@ class SchemaUnmarshallersFactory(object):
context=self.context,
)
formatter = self.get_formatter(klass.FORMATTERS, schema.format)
schema_format = schema.getkey('format')
formatter = self.get_formatter(klass.FORMATTERS, schema_format)
if formatter is None:
raise FormatterNotFoundError(schema.format)
raise FormatterNotFoundError(schema_format)
validator = self.get_validator(schema)
return klass(formatter, validator, **kwargs)
def get_formatter(self, default_formatters, type_format=SchemaFormat.NONE):
def get_formatter(self, default_formatters, type_format=None):
try:
schema_format = SchemaFormat(type_format)
except ValueError:
return self.custom_formatters.get(type_format)
else:
return default_formatters.get(schema_format)
return self.custom_formatters[type_format]
except KeyError:
return default_formatters.get(type_format)
def get_validator(self, schema):
kwargs = {
@ -87,4 +85,5 @@ class SchemaUnmarshallersFactory(object):
}
if self.context is not None:
kwargs[self.CONTEXT_VALIDATION[self.context]] = True
return OAS30Validator(schema.__dict__, **kwargs)
with schema.open() as schema_dict:
return OAS30Validator(schema_dict, **kwargs)

113
openapi_core/unmarshalling/schemas/unmarshallers.py
View file

@ -1,3 +1,4 @@
from __future__ import division
from functools import partial
import logging
@ -12,9 +13,10 @@ from six import text_type, binary_type
from six import iteritems
from openapi_core.extensions.models.factories import ModelFactory
from openapi_core.schema.schemas.enums import SchemaFormat, SchemaType
from openapi_core.schema.schemas.models import Schema
from openapi_core.schema.schemas.types import NoValue
from openapi_core.schema.schemas import (
get_all_properties, get_all_properties_names
)
from openapi_core.types import NoValue
from openapi_core.unmarshalling.schemas.enums import UnmarshalContext
from openapi_core.unmarshalling.schemas.exceptions import (
UnmarshalError, ValidateError, InvalidSchemaValue,
@ -40,7 +42,7 @@ class PrimitiveTypeUnmarshaller(object):
def __call__(self, value=NoValue):
if value is NoValue:
value = self.schema.default
value = self.schema.getkey('default')
if value is None:
return
@ -51,40 +53,43 @@ class PrimitiveTypeUnmarshaller(object):
def _formatter_validate(self, value):
result = self.formatter.validate(value)
if not result:
raise InvalidSchemaValue(value, self.schema.type)
schema_type = self.schema.getkey('type', 'any')
raise InvalidSchemaValue(value, schema_type)
def validate(self, value):
errors_iter = self.validator.iter_errors(value)
errors = tuple(errors_iter)
if errors:
schema_type = self.schema.getkey('type', 'any')
raise InvalidSchemaValue(
value, self.schema.type, schema_errors=errors)
value, schema_type, schema_errors=errors)
def unmarshal(self, value):
try:
return self.formatter.unmarshal(value)
except ValueError as exc:
schema_format = self.schema.getkey('format')
raise InvalidSchemaFormatValue(
value, self.schema.format, exc)
value, schema_format, exc)
class StringUnmarshaller(PrimitiveTypeUnmarshaller):
FORMATTERS = {
SchemaFormat.NONE: Formatter.from_callables(
None: Formatter.from_callables(
partial(is_string, None), text_type),
SchemaFormat.PASSWORD: Formatter.from_callables(
'password': Formatter.from_callables(
partial(oas30_format_checker.check, format='password'), text_type),
SchemaFormat.DATE: Formatter.from_callables(
'date': Formatter.from_callables(
partial(oas30_format_checker.check, format='date'), format_date),
SchemaFormat.DATETIME: Formatter.from_callables(
'date-time': Formatter.from_callables(
partial(oas30_format_checker.check, format='date-time'),
parse_datetime),
SchemaFormat.BINARY: Formatter.from_callables(
'binary': Formatter.from_callables(
partial(oas30_format_checker.check, format='binary'), binary_type),
SchemaFormat.UUID: Formatter.from_callables(
'uuid': Formatter.from_callables(
partial(oas30_format_checker.check, format='uuid'), format_uuid),
SchemaFormat.BYTE: Formatter.from_callables(
'byte': Formatter.from_callables(
partial(oas30_format_checker.check, format='byte'), format_byte),
}
@ -92,11 +97,11 @@ class StringUnmarshaller(PrimitiveTypeUnmarshaller):
class IntegerUnmarshaller(PrimitiveTypeUnmarshaller):
FORMATTERS = {
SchemaFormat.NONE: Formatter.from_callables(
None: Formatter.from_callables(
partial(is_integer, None), int),
SchemaFormat.INT32: Formatter.from_callables(
'int32': Formatter.from_callables(
partial(oas30_format_checker.check, format='int32'), int),
SchemaFormat.INT64: Formatter.from_callables(
'int64': Formatter.from_callables(
partial(oas30_format_checker.check, format='int64'), int),
}
@ -104,11 +109,11 @@ class IntegerUnmarshaller(PrimitiveTypeUnmarshaller):
class NumberUnmarshaller(PrimitiveTypeUnmarshaller):
FORMATTERS = {
SchemaFormat.NONE: Formatter.from_callables(
None: Formatter.from_callables(
partial(is_number, None), format_number),
SchemaFormat.FLOAT: Formatter.from_callables(
'float': Formatter.from_callables(
partial(oas30_format_checker.check, format='float'), float),
SchemaFormat.DOUBLE: Formatter.from_callables(
'double': Formatter.from_callables(
partial(oas30_format_checker.check, format='double'), float),
}
@ -116,7 +121,7 @@ class NumberUnmarshaller(PrimitiveTypeUnmarshaller):
class BooleanUnmarshaller(PrimitiveTypeUnmarshaller):
FORMATTERS = {
SchemaFormat.NONE: Formatter.from_callables(
None: Formatter.from_callables(
partial(is_bool, None), forcebool),
}
@ -134,17 +139,17 @@ class ComplexUnmarshaller(PrimitiveTypeUnmarshaller):
class ArrayUnmarshaller(ComplexUnmarshaller):
FORMATTERS = {
SchemaFormat.NONE: Formatter.from_callables(
None: Formatter.from_callables(
partial(is_array, None), list),
}
@property
def items_unmarshaller(self):
return self.unmarshallers_factory.create(self.schema.items)
return self.unmarshallers_factory.create(self.schema / 'items')
def __call__(self, value=NoValue):
value = super(ArrayUnmarshaller, self).__call__(value)
if value is None and self.schema.nullable:
if value is None and self.schema.getkey('nullable', False):
return None
return list(map(self.items_unmarshaller, value))
@ -152,7 +157,7 @@ class ArrayUnmarshaller(ComplexUnmarshaller):
class ObjectUnmarshaller(ComplexUnmarshaller):
FORMATTERS = {
SchemaFormat.NONE: Formatter.from_callables(
None: Formatter.from_callables(
partial(is_object, None), dict),
}
@ -170,9 +175,9 @@ class ObjectUnmarshaller(ComplexUnmarshaller):
return self._unmarshal_object(value)
def _unmarshal_object(self, value=NoValue):
if self.schema.one_of:
if 'oneOf' in self.schema:
properties = None
for one_of_schema in self.schema.one_of:
for one_of_schema in self.schema / 'oneOf':
try:
unmarshalled = self._unmarshal_properties(
value, one_of_schema)
@ -190,46 +195,50 @@ class ObjectUnmarshaller(ComplexUnmarshaller):
else:
properties = self._unmarshal_properties(value)
if 'x-model' in self.schema.extensions:
extension = self.schema.extensions['x-model']
return self.model_factory.create(properties, name=extension.value)
if 'x-model' in self.schema:
name = self.schema['x-model']
return self.model_factory.create(properties, name=name)
return properties
def _unmarshal_properties(self, value=NoValue, one_of_schema=None):
all_props = self.schema.get_all_properties()
all_props_names = self.schema.get_all_properties_names()
all_props = get_all_properties(self.schema)
all_props_names = get_all_properties_names(self.schema)
if one_of_schema is not None:
all_props.update(one_of_schema.get_all_properties())
all_props_names |= one_of_schema.\
get_all_properties_names()
all_props.update(get_all_properties(one_of_schema))
all_props_names |= get_all_properties_names(one_of_schema)
value_props_names = value.keys()
extra_props = set(value_props_names) - set(all_props_names)
properties = {}
if isinstance(self.schema.additional_properties, Schema):
additional_properties = self.schema.getkey(
'additionalProperties', True)
if isinstance(additional_properties, dict):
additional_prop_schema = self.schema / 'additionalProperties'
for prop_name in extra_props:
prop_value = value[prop_name]
properties[prop_name] = self.unmarshallers_factory.create(
self.schema.additional_properties)(prop_value)
elif self.schema.additional_properties is True:
additional_prop_schema)(prop_value)
elif additional_properties is True:
for prop_name in extra_props:
prop_value = value[prop_name]
properties[prop_name] = prop_value
for prop_name, prop in iteritems(all_props):
if self.context == UnmarshalContext.REQUEST and prop.read_only:
read_only = prop.getkey('readOnly', False)
if self.context == UnmarshalContext.REQUEST and read_only:
continue
if self.context == UnmarshalContext.RESPONSE and prop.write_only:
write_only = prop.getkey('writeOnly', False)
if self.context == UnmarshalContext.RESPONSE and write_only:
continue
try:
prop_value = value[prop_name]
except KeyError:
if prop.default is NoValue:
if 'default' not in prop:
continue
prop_value = prop.default
prop_value = prop['default']
properties[prop_name] = self.unmarshallers_factory.create(
prop)(prop_value)
@ -240,12 +249,12 @@ class ObjectUnmarshaller(ComplexUnmarshaller):
class AnyUnmarshaller(ComplexUnmarshaller):
FORMATTERS = {
SchemaFormat.NONE: Formatter(),
None: Formatter(),
}
SCHEMA_TYPES_ORDER = [
SchemaType.OBJECT, SchemaType.ARRAY, SchemaType.BOOLEAN,
SchemaType.INTEGER, SchemaType.NUMBER, SchemaType.STRING,
'object', 'array', 'boolean',
'integer', 'number', 'string',
]
def unmarshal(self, value=NoValue):
@ -272,9 +281,11 @@ class AnyUnmarshaller(ComplexUnmarshaller):
return value
def _get_one_of_schema(self, value):
if not self.schema.one_of:
if 'oneOf' not in self.schema:
return
for subschema in self.schema.one_of:
one_of_schemas = self.schema / 'oneOf'
for subschema in one_of_schemas:
unmarshaller = self.unmarshallers_factory.create(subschema)
try:
unmarshaller.validate(value)
@ -284,10 +295,12 @@ class AnyUnmarshaller(ComplexUnmarshaller):
return subschema
def _get_all_of_schema(self, value):
if not self.schema.all_of:
if 'allOf' not in self.schema:
return
for subschema in self.schema.all_of:
if subschema.type == SchemaType.ANY:
all_of_schemas = self.schema / 'allOf'
for subschema in all_of_schemas:
if 'type' not in subschema:
continue
unmarshaller = self.unmarshallers_factory.create(subschema)
try:

13
openapi_core/validation/request/shortcuts.py
View file

@ -1,21 +1,8 @@
"""OpenAPI core validation request shortcuts module"""
import warnings
from openapi_core.schema.media_types.exceptions import OpenAPIMediaTypeError
from openapi_core.schema.parameters.exceptions import OpenAPIParameterError
from openapi_core.schema.request_bodies.exceptions import (
OpenAPIRequestBodyError,
)
from openapi_core.schema.schemas.exceptions import OpenAPISchemaError
from openapi_core.validation.request.validators import RequestValidator
ERRORS_BODY = (
OpenAPIRequestBodyError, OpenAPIMediaTypeError, OpenAPISchemaError,
)
ERRORS_PARAMETERS = (
OpenAPIParameterError,
)
def validate_request(validator, request, failsafe=None):
if failsafe is None:

100
openapi_core/validation/request/validators.py
View file

@ -1,14 +1,14 @@
"""OpenAPI core validation request validators module"""
from __future__ import division
from itertools import chain
from six import iteritems
from openapi_core.casting.schemas.exceptions import CastError
from openapi_core.deserializing.exceptions import DeserializeError
from openapi_core.schema.parameters.exceptions import (
MissingRequiredParameter, MissingParameter,
from openapi_core.exceptions import (
MissingRequiredParameter, MissingParameter, MissingRequestBody,
)
from openapi_core.schema.request_bodies.exceptions import MissingRequestBody
from openapi_core.security.exceptions import SecurityError
from openapi_core.schema.parameters import get_aslist, get_explode
from openapi_core.templating.media_types.exceptions import MediaTypeFinderError
from openapi_core.templating.paths.exceptions import PathError
from openapi_core.unmarshalling.schemas.enums import UnmarshalContext
@ -38,10 +38,17 @@ class RequestValidator(BaseValidator):
request.parameters.path = request.parameters.path or \
path_result.variables
operation_params = operation.get('parameters', [])
operation_params_iter = operation_params and \
iter(operation_params) or []
path_params = path.get('parameters', [])
params_params_iter = path_params and \
iter(path_params) or []
params, params_errors = self._get_parameters(
request, chain(
iteritems(operation.parameters),
iteritems(path.parameters)
operation_params_iter,
params_params_iter,
)
)
@ -63,10 +70,17 @@ class RequestValidator(BaseValidator):
request.parameters.path = request.parameters.path or \
path_result.variables
operation_params = operation.get('parameters', [])
operation_params_iter = operation_params and \
iter(operation_params) or []
path_params = path.get('parameters', [])
params_params_iter = path_params and \
iter(path_params) or []
params, params_errors = self._get_parameters(
request, chain(
iteritems(operation.parameters),
iteritems(path.parameters)
operation_params_iter,
params_params_iter,
)
)
return RequestValidationResult(
@ -87,9 +101,11 @@ class RequestValidator(BaseValidator):
)
def _get_security(self, request, operation):
security = self.spec.security
if operation.security is not None:
security = operation.security
security = None
if 'security' in self.spec:
security = self.spec / 'security'
if 'security' in operation:
security = operation / 'security'
if not security:
return {}
@ -99,7 +115,7 @@ class RequestValidator(BaseValidator):
return {
scheme_name: self._get_security_value(
scheme_name, request)
for scheme_name in security_requirement
for scheme_name in security_requirement.keys()
}
except SecurityError:
continue
@ -110,21 +126,26 @@ class RequestValidator(BaseValidator):
errors = []
seen = set()
locations = {}
for param_name, param in params:
if (param_name, param.location.value) in seen:
for param in params:
param_name = param['name']
param_location = param['in']
if (param_name, param_location) in seen:
# skip parameter already seen
# e.g. overriden path item paremeter on operation
continue
seen.add((param_name, param.location.value))
seen.add((param_name, param_location))
try:
raw_value = self._get_parameter_value(param, request)
except MissingRequiredParameter as exc:
errors.append(exc)
continue
except MissingParameter:
if not param.schema or not param.schema.has_default():
if 'schema' not in param:
continue
casted = param.schema.default
schema = param / 'schema'
if 'default' not in schema:
continue
casted = schema['default']
else:
try:
deserialised = self._deserialise_parameter(
@ -144,28 +165,29 @@ class RequestValidator(BaseValidator):
except (ValidateError, UnmarshalError) as exc:
errors.append(exc)
else:
locations.setdefault(param.location.value, {})
locations[param.location.value][param_name] = unmarshalled
locations.setdefault(param_location, {})
locations[param_location][param_name] = unmarshalled
return RequestParameters(**locations), errors
def _get_body(self, request, operation):
if operation.request_body is None:
if 'requestBody' not in operation:
return None, []
request_body = operation / 'requestBody'
try:
media_type = self._get_media_type(
operation.request_body.content, request)
media_type, mimetype = self._get_media_type(
request_body / 'content', request)
except MediaTypeFinderError as exc:
return None, [exc, ]
try:
raw_body = self._get_body_value(operation.request_body, request)
raw_body = self._get_body_value(request_body, request)
except MissingRequestBody as exc:
return None, [exc, ]
try:
deserialised = self._deserialise_media_type(media_type, raw_body)
deserialised = self._deserialise_data(mimetype, raw_body)
except DeserializeError as exc:
return None, [exc, ]
@ -182,33 +204,37 @@ class RequestValidator(BaseValidator):
return body, []
def _get_security_value(self, scheme_name, request):
scheme = self.spec.components.security_schemes.get(scheme_name)
if not scheme:
security_schemes = self.spec / 'components#securitySchemes'
if scheme_name not in security_schemes:
return
scheme = security_schemes[scheme_name]
from openapi_core.security.factories import SecurityProviderFactory
security_provider_factory = SecurityProviderFactory()
security_provider = security_provider_factory.create(scheme)
return security_provider(request)
def _get_parameter_value(self, param, request):
location = request.parameters[param.location.value]
param_location = param['in']
location = request.parameters[param_location]
if param.name not in location:
if param.required:
raise MissingRequiredParameter(param.name)
if param['name'] not in location:
if param.getkey('required', False):
raise MissingRequiredParameter(param['name'])
raise MissingParameter(param.name)
raise MissingParameter(param['name'])
if param.aslist and param.explode:
aslist = get_aslist(param)
explode = get_explode(param)
if aslist and explode:
if hasattr(location, 'getall'):
return location.getall(param.name)
return location.getlist(param.name)
return location.getall(param['name'])
return location.getlist(param['name'])
return location[param.name]
return location[param['name']]
def _get_body_value(self, request_body, request):
if not request.body and request_body.required:
required = request_body.getkey('required', False)
if not request.body and required:
raise MissingRequestBody(request)
return request.body

Some files were not shown because too many files have changed in this diff Show more