from functools import partial import logging 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_datetime, 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'), format_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