from functools import partial
import logging
from isodate.isodatetime import parse_datetime
from openapi_schema_validator._types import (
is_array, is_bool, is_integer,
is_object, is_number, is_string,
)
from openapi_schema_validator._format import oas30_format_checker
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.unmarshalling.schemas.enums import UnmarshalContext
from openapi_core.unmarshalling.schemas.exceptions import (
UnmarshalError, ValidateError, InvalidSchemaValue,
InvalidSchemaFormatValue,
)
from openapi_core.unmarshalling.schemas.formatters import Formatter
from openapi_core.unmarshalling.schemas.util import (
forcebool, format_date, format_byte, format_uuid,
format_number,
)
log = logging.getLogger(__name__)
class PrimitiveTypeUnmarshaller(object):
FORMATTERS = {}
def __init__(self, formatter, validator, schema):
self.formatter = formatter
self.validator = validator
self.schema = schema
def __call__(self, value=NoValue):
if value is NoValue:
value = self.schema.default
if value is None:
return
self.validate(value)
return self.unmarshal(value)
def _formatter_validate(self, value):
result = self.formatter.validate(value)
if not result:
raise InvalidSchemaValue(value, self.schema.type)
def validate(self, value):
errors_iter = self.validator.iter_errors(value)
errors = tuple(errors_iter)
if errors:
raise InvalidSchemaValue(
value, self.schema.type, schema_errors=errors)
def unmarshal(self, value):
try:
return self.formatter.unmarshal(value)
except ValueError as exc:
raise InvalidSchemaFormatValue(
value, self.schema.format, exc)
class StringUnmarshaller(PrimitiveTypeUnmarshaller):
FORMATTERS = {
SchemaFormat.NONE: Formatter.from_callables(
partial(is_string, None), text_type),
SchemaFormat.PASSWORD: Formatter.from_callables(
partial(oas30_format_checker.check, format='password'), text_type),
SchemaFormat.DATE: Formatter.from_callables(
partial(oas30_format_checker.check, format='date'), format_date),
SchemaFormat.DATETIME: Formatter.from_callables(
partial(oas30_format_checker.check, format='date-time'),
parse_datetime),
SchemaFormat.BINARY: Formatter.from_callables(
partial(oas30_format_checker.check, format='binary'), binary_type),
SchemaFormat.UUID: Formatter.from_callables(
partial(oas30_format_checker.check, format='uuid'), format_uuid),
SchemaFormat.BYTE: Formatter.from_callables(
partial(oas30_format_checker.check, format='byte'), format_byte),
}
class IntegerUnmarshaller(PrimitiveTypeUnmarshaller):
FORMATTERS = {
SchemaFormat.NONE: Formatter.from_callables(
partial(is_integer, None), int),
SchemaFormat.INT32: Formatter.from_callables(
partial(oas30_format_checker.check, format='int32'), int),
SchemaFormat.INT64: Formatter.from_callables(
partial(oas30_format_checker.check, format='int64'), int),
}
class NumberUnmarshaller(PrimitiveTypeUnmarshaller):
FORMATTERS = {
SchemaFormat.NONE: Formatter.from_callables(
partial(is_number, None), format_number),
SchemaFormat.FLOAT: Formatter.from_callables(
partial(oas30_format_checker.check, format='float'), float),
SchemaFormat.DOUBLE: Formatter.from_callables(
partial(oas30_format_checker.check, format='double'), float),
}
class BooleanUnmarshaller(PrimitiveTypeUnmarshaller):
FORMATTERS = {
SchemaFormat.NONE: Formatter.from_callables(
partial(is_bool, None), forcebool),
}
class ComplexUnmarshaller(PrimitiveTypeUnmarshaller):
def __init__(
self, formatter, validator, schema, unmarshallers_factory,
context=None):
super(ComplexUnmarshaller, self).__init__(formatter, validator, schema)
self.unmarshallers_factory = unmarshallers_factory
self.context = context
class ArrayUnmarshaller(ComplexUnmarshaller):
FORMATTERS = {
SchemaFormat.NONE: Formatter.from_callables(
partial(is_array, None), list),
}
@property
def items_unmarshaller(self):
return self.unmarshallers_factory.create(self.schema.items)
def __call__(self, value=NoValue):
value = super(ArrayUnmarshaller, self).__call__(value)
return list(map(self.items_unmarshaller, value))
class ObjectUnmarshaller(ComplexUnmarshaller):
FORMATTERS = {
SchemaFormat.NONE: Formatter.from_callables(
partial(is_object, None), dict),
}
@property
def model_factory(self):
return ModelFactory()
def unmarshal(self, value):
try:
value = self.formatter.unmarshal(value)
except ValueError as exc:
raise InvalidSchemaFormatValue(
value, self.schema.format, exc)
else:
return self._unmarshal_object(value)
def _unmarshal_object(self, value=NoValue):
if self.schema.one_of:
properties = None
for one_of_schema in self.schema.one_of:
try:
unmarshalled = self._unmarshal_properties(
value, one_of_schema)
except (UnmarshalError, ValueError):
pass
else:
if properties is not None:
log.warning("multiple valid oneOf schemas found")
continue
properties = unmarshalled
if properties is None:
log.warning("valid oneOf schema not found")
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)
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()
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()
value_props_names = value.keys()
extra_props = set(value_props_names) - set(all_props_names)
properties = {}
if isinstance(self.schema.additional_properties, Schema):
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:
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:
continue
if self.context == UnmarshalContext.RESPONSE and prop.write_only:
continue
try:
prop_value = value[prop_name]
except KeyError:
if prop.default is NoValue:
continue
prop_value = prop.default
properties[prop_name] = self.unmarshallers_factory.create(
prop)(prop_value)
return properties
class AnyUnmarshaller(ComplexUnmarshaller):
FORMATTERS = {
SchemaFormat.NONE: Formatter(),
}
SCHEMA_TYPES_ORDER = [
SchemaType.OBJECT, SchemaType.ARRAY, SchemaType.BOOLEAN,
SchemaType.INTEGER, SchemaType.NUMBER, SchemaType.STRING,
]
def __call__(self, value=NoValue):
one_of_schema = self._get_one_of_schema(value)
if one_of_schema:
return self.unmarshallers_factory.create(one_of_schema)(value)
for schema_type in self.SCHEMA_TYPES_ORDER:
unmarshaller = self.unmarshallers_factory.create(
self.schema, type_override=schema_type)
# validate with validator of formatter (usualy type validator)
try:
unmarshaller._formatter_validate(value)
except ValidateError:
continue
else:
return unmarshaller(value)
log.warning("failed to unmarshal any type")
return value
def _get_one_of_schema(self, value):
if not self.schema.one_of:
return
for subschema in self.schema.one_of:
unmarshaller = self.unmarshallers_factory.create(subschema)
try:
unmarshaller.validate(value)
except ValidateError:
continue
else:
return subschema