Source code for transmute_core.object_serializers.schematics_serializer

from .interface import ObjectSerializer
from collections import OrderedDict
from datetime import datetime
from schematics.types import (
from schematics.models import ModelMeta, Model
from schematics.types.compound import ListType, ModelType, DictType
from schematics.exceptions import BaseError, ConversionError
from schematics.transforms import get_import_context
from ..exceptions import SerializationException
from decimal import Decimal
from ..compat import all_string_types

JSON_SCHEMA_MAP = OrderedDict(
        (BooleanType, {"type": "boolean"}),
        (IntType, {"type": "integer"}),
        (NumberType, {"type": "number"}),
        (UUIDType, {"type": "string", "format": "uuid"}),
        (URLType, {"type": "string", "format": "url"}),
        (StringType, {"type": "string"}),
        (DateTimeType, {"type": "string", "format": "date-time"}),
        (BaseType, {"type": "object"}),

[docs]class SchematicsSerializer(ObjectSerializer): """ An ObjectSerializer which allows the serialization of basic types and schematics models. The valid types that SchematicsSerializer supports are: - int - float - bool - decimal - string - none - lists, in the form of [Type] (e.g. [str]) - any type that extends the schematics.models.Model. """ VALID_BASE_CLASSES = [BaseType, ModelMeta, Model, Serializable] def can_handle(self, cls): if cls in self._models: return True if any(isinstance(cls, t) for t in self.VALID_BASE_CLASSES): return True if not isinstance(cls, type): return False if any(issubclass(cls, t) for t in self.VALID_BASE_CLASSES): return True def __init__(self, builtin_models=None): self._models = {} def _translate_to_model(self, model): model = self._to_key(model) if isinstance(model, BaseType): return model if isinstance(model, ModelMeta): return ModelType(model) if model not in self._models and isinstance(model, tuple): self._models[model] = ListType(self._translate_to_model(model[0])) if model in self._models: return self._models[model] return model @staticmethod def _to_key(model): if isinstance(model, list): return tuple(model) return model
[docs] def load(self, model, value): model = _enforce_instance(model) try: context = get_import_context(oo=True) model = self._translate_to_model(model) result = model(value, context=context) if isinstance(model, ModelType) or not isinstance(model, BaseType): result.validate() else: model.validate(result, context=context) return result except BaseError as e: raise SerializationException(str(e)) except ConversionError as e: raise SerializationException(str(e))
[docs] def dump(self, model, value): model = _enforce_instance(model) try: model = self._translate_to_model(model) return model.to_primitive(value) except BaseError as e: raise SerializationException(str(e))
[docs] def to_json_schema(self, model): if isinstance(model, Serializable): model = model.type model = _enforce_instance(model) model = self._translate_to_model(model) return _to_json_schema(model)
_cache = {} def _to_json_schema(model): if model not in _cache: _cache[model] = _to_json_schema_no_cache(model) return _cache[model] def _to_json_schema_no_cache(model): if isinstance(model, ModelType): return _model_type_to_json_schema(model) elif isinstance(model, ListType): return _list_type_to_json_schema(model) elif isinstance(model, DictType): return _dict_type_to_json_schema(model) else: for cls, schema in JSON_SCHEMA_MAP.items(): if isinstance(model, cls): return schema if isinstance(model, BaseType): return {"type": "object"} if isinstance(model, Serializable): return _to_json_schema_no_cache(model.type) raise SerializationException( "unable to create json schema for type " + "{0}. Expected a primitive or ".format(model) + " a schematics model or type" ) def _model_type_to_json_schema(model): required = [] schema = {"title": model.model_name, "type": "object", "properties": {}} for name, field in model.fields.items(): if field.required: required.append(name) schema["properties"][name] = _to_json_schema(field) if len(required) > 0: schema["required"] = required return schema def _list_type_to_json_schema(list_type): return {"type": "array", "items": _to_json_schema(list_type.field)} def _dict_type_to_json_schema(dict_type): return {"type": "object", "additionalProperties": _to_json_schema(dict_type.field)} def _enforce_instance(model_or_class): """ It's a common mistake to not initialize a schematics class. We should handle that by just calling the default constructor. """ if isinstance(model_or_class, type) and issubclass(model_or_class, BaseType): return model_or_class() return model_or_class