jsons._main_impl.set_deserializer() - Code Metrics - Inspection of "Release/0.8.0" - ramonhagenaars/jsons - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#23)

by Ramon

created 2019-03-17 10:56 UTC

jsons._main_impl.set_deserializer() A

↳ Parent: jsons._main_impl

Complexity

Conditions

Size

Total Lines	30
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	10
nop	4
dl	0
loc	30
rs	9.9
c	0
b	0
f	0

"""
PRIVATE MODULE: do not import (from) it directly.

This module contains the implementation of the main functions of jsons, such
as `load` and `dump`.
"""
import json
from importlib import import_module
from json import JSONDecodeError
from typing import Dict, Callable, Optional, Union, Tuple
from jsons._common_impl import (
    get_class_name,
    get_parents,
    META_ATTR,
    StateHolder
)
from jsons.exceptions import (
    DecodeError,
    DeserializationError,
    JsonsError,
    SerializationError,
    UnknownClassError
)

VALID_TYPES = (str, int, float, bool, list, tuple, set, dict, type(None))
RFC3339_DATETIME_PATTERN = '%Y-%m-%dT%H:%M:%S'


def dump(obj: object,
         cls: Optional[type] = None,
         fork_inst: Optional[type] = StateHolder,
         **kwargs) -> object:
    """
    Serialize the given ``obj`` to a JSON equivalent type (e.g. dict, list,
    int, ...).

    The way objects are serialized can be finetuned by setting serializer
    functions for the specific type using ``set_serializer``.

    You can also provide ``cls`` to specify that ``obj`` needs to be serialized
    as if it was of type ``cls`` (meaning to only take into account attributes
    from ``cls``). The type ``cls`` must have a ``__slots__`` defined. Any type
    will do, but in most cases you may want ``cls`` to be a base class of
    ``obj``.
    :param obj: a Python instance of any sort.
    :param cls: if given, ``obj`` will be dumped as if it is of type ``type``.
    :param fork_inst: if given, it uses this fork of ``JsonSerializable``.
    :param kwargs: the keyword args are passed on to the serializer function.
    :return: the serialized obj as a JSON type.
    """
    if cls and not hasattr(cls, '__slots__'):
        raise SerializationError('Invalid type: "{}". Only types that have a '
                                 '__slots__ defined are allowed when '
                                 'providing "cls".'
                         .format(get_class_name(cls, fork_inst=fork_inst)))
    cls_ = cls or obj.__class__
    serializer = _get_serializer(cls_, fork_inst)
    kwargs_ = {
        'fork_inst': fork_inst,
        **kwargs
    }
    try:
        return serializer(obj, cls=cls, **kwargs_)
    except Exception as err:
        raise SerializationError(str(err))


def load(json_obj: object,
         cls: Optional[type] = None,
         strict: bool = False,
         fork_inst: Optional[type] = StateHolder,
         attr_getters: Optional[Dict[str, Callable[[], object]]] = None,
         **kwargs) -> object:
    """
    Deserialize the given ``json_obj`` to an object of type ``cls``. If the
    contents of ``json_obj`` do not match the interface of ``cls``, a
    DeserializationError is raised.

    If ``json_obj`` contains a value that belongs to a custom class, there must
    be a type hint present for that value in ``cls`` to let this function know
    what type it should deserialize that value to.


    **Example**:

    >>> from typing import List
    >>> import jsons
    >>> class Person:
    ...     # No type hint required for name
    ...     def __init__(self, name):
    ...         self.name = name
    >>> class Family:
    ...     # Person is a custom class, use a type hint
    ...         def __init__(self, persons: List[Person]):
    ...             self.persons = persons
    >>> loaded = jsons.load({'persons': [{'name': 'John'}]}, Family)
    >>> loaded.persons[0].name
    'John'

    If no ``cls`` is given, a dict is simply returned, but contained values
    (e.g. serialized ``datetime`` values) are still deserialized.

    If `strict` mode is off and the type of `json_obj` exactly matches `cls`
    then `json_obj` is simply returned.

    :param json_obj: the dict that is to be deserialized.
    :param cls: a matching class of which an instance should be returned.
    :param strict: a bool to determine if the deserializer should be strict
    (i.e. fail on a partially deserialized `json_obj` or on `None`).
    :param fork_inst: if given, it uses this fork of ``JsonSerializable``.
    :param attr_getters: a ``dict`` that may hold callables that return values
    for certain attributes.
    :param kwargs: the keyword args are passed on to the deserializer function.
    :return: an instance of ``cls`` if given, a dict otherwise.
    """
    if not strict and (json_obj is None or type(json_obj) == cls):
        return json_obj
    cls = _check_and_get_cls(json_obj, cls, fork_inst)
    deserializer = _get_deserializer(cls, fork_inst)
    kwargs_ = {
        'strict': strict,
        'fork_inst': fork_inst,
        'attr_getters': attr_getters,
        **kwargs
    }
    try:
        return deserializer(json_obj, cls, **kwargs_)
    except Exception as err:
        if isinstance(err, JsonsError):
            raise
        raise DeserializationError(str(err), json_obj, cls)


def _get_serializer(cls: type,
                    fork_inst: Optional[type] = StateHolder) -> callable:
    serializer = _get_lizer(cls, fork_inst._serializers,
                            fork_inst._classes_serializers, fork_inst)
    return serializer


def _get_deserializer(cls: type,
                      fork_inst: Optional[type] = StateHolder) -> callable:
    deserializer = _get_lizer(cls, fork_inst._deserializers,
                              fork_inst._classes_deserializers, fork_inst)
    return deserializer


def _get_lizer(cls: type,
               lizers: Dict[str, callable],
               classes_lizers: list,
               fork_inst: type) -> callable:
    cls_name = get_class_name(cls, str.lower, fork_inst=fork_inst)
    lizer = lizers.get(cls_name, None)
    if not lizer:
        parents = get_parents(cls, classes_lizers)
        if parents:
            pname = get_class_name(parents[0], str.lower, fork_inst=fork_inst)
            lizer = lizers[pname]
    return lizer


def dumps(obj: object,
          jdkwargs: Optional[Dict[str, object]] = None,
          *args,
          **kwargs) -> str:
    """
    Extend ``json.dumps``, allowing any Python instance to be dumped to a
    string. Any extra (keyword) arguments are passed on to ``json.dumps``.

    :param obj: the object that is to be dumped to a string.
    :param jdkwargs: extra keyword arguments for ``json.dumps`` (not
    ``jsons.dumps``!)
    :param args: extra arguments for ``jsons.dumps``.
    :param kwargs: Keyword arguments that are passed on through the
    serialization process.
    passed on to the serializer function.
    :return: ``obj`` as a ``str``.
    """
    jdkwargs = jdkwargs or {}
    dumped = dump(obj, *args, **kwargs)
    return json.dumps(dumped, **jdkwargs)


def loads(str_: str,
          cls: Optional[type] = None,
          jdkwargs: Optional[Dict[str, object]] = None,
          *args,
          **kwargs) -> object:
    """
    Extend ``json.loads``, allowing a string to be loaded into a dict or a
    Python instance of type ``cls``. Any extra (keyword) arguments are passed
    on to ``json.loads``.

    :param str_: the string that is to be loaded.
    :param cls: a matching class of which an instance should be returned.
    :param jdkwargs: extra keyword arguments for ``json.loads`` (not
    ``jsons.loads``!)
    :param args: extra arguments for ``jsons.loads``.
    :param kwargs: extra keyword arguments for ``jsons.loads``.
    :return: a JSON-type object (dict, str, list, etc.) or an instance of type
    ``cls`` if given.
    """
    jdkwargs = jdkwargs or {}
    try:
        obj = json.loads(str_, **jdkwargs)
    except JSONDecodeError as err:
        raise DecodeError('Could not load a dict; the given string is not '
                          'valid JSON.', str_, cls, err)
    else:
        return load(obj, cls, *args, **kwargs)


def dumpb(obj: object,
          encoding: str = 'utf-8',
          jdkwargs: Optional[Dict[str, object]] = None,
          *args,
          **kwargs) -> bytes:
    """
    Extend ``json.dumps``, allowing any Python instance to be dumped to bytes.
    Any extra (keyword) arguments are passed on to ``json.dumps``.

    :param obj: the object that is to be dumped to bytes.
    :param encoding: the encoding that is used to transform to bytes.
    :param jdkwargs: extra keyword arguments for ``json.dumps`` (not
    ``jsons.dumps``!)
    :param args: extra arguments for ``jsons.dumps``.
    :param kwargs: Keyword arguments that are passed on through the
    serialization process.
    passed on to the serializer function.
    :return: ``obj`` as ``bytes``.
    """
    jdkwargs = jdkwargs or {}
    dumped_dict = dump(obj, *args, **kwargs)
    dumped_str = json.dumps(dumped_dict, **jdkwargs)
    return dumped_str.encode(encoding=encoding)


def loadb(bytes_: bytes,
          cls: Optional[type] = None,
          encoding: str = 'utf-8',
          jdkwargs: Optional[Dict[str, object]] = None,
          *args,
          **kwargs) -> object:
    """
    Extend ``json.loads``, allowing bytes to be loaded into a dict or a Python
    instance of type ``cls``. Any extra (keyword) arguments are passed on to
    ``json.loads``.

    :param bytes_: the bytes that are to be loaded.
    :param cls: a matching class of which an instance should be returned.
    :param encoding: the encoding that is used to transform from bytes.
    :param jdkwargs: extra keyword arguments for ``json.loads`` (not
    ``jsons.loads``!)
    :param args: extra arguments for ``jsons.loads``.
    :param kwargs: extra keyword arguments for ``jsons.loads``.
    :return: a JSON-type object (dict, str, list, etc.) or an instance of type
    ``cls`` if given.
    """
    if not isinstance(bytes_, bytes):
        raise DeserializationError('loadb accepts bytes only, "{}" was given'
                                   .format(type(bytes_)), bytes_, cls)
    jdkwargs = jdkwargs or {}
    str_ = bytes_.decode(encoding=encoding)
    return loads(str_, cls, jdkwargs=jdkwargs, *args, **kwargs)


def set_serializer(func: callable,
                   cls: type,
                   high_prio: bool = True,
                   fork_inst: type = StateHolder) -> None:
    """
    Set a serializer function for the given type. You may override the default
    behavior of ``jsons.load`` by setting a custom serializer.

    The ``func`` argument must take one argument (i.e. the object that is to be
    serialized) and also a ``kwargs`` parameter. For example:

    >>> def func(obj, **kwargs):
    ...    return dict()

    You may ask additional arguments between ``cls`` and ``kwargs``.

    :param func: the serializer function.
    :param cls: the type this serializer can handle.
    :param high_prio: determines the order in which is looked for the callable.
    :param fork_inst: if given, it uses this fork of ``JsonSerializable``.
    :return: None.
    """
    if cls:
        index = 0 if high_prio else len(fork_inst._classes_serializers)
        fork_inst._classes_serializers.insert(index, cls)
        cls_name = get_class_name(cls, fork_inst=fork_inst)
        fork_inst._serializers[cls_name.lower()] = func
    else:
        fork_inst._serializers['nonetype'] = func


def set_deserializer(func: callable,
                     cls: Union[type, str],
                     high_prio: bool = True,
                     fork_inst: type = StateHolder) -> None:
    """
    Set a deserializer function for the given type. You may override the
    default behavior of ``jsons.dump`` by setting a custom deserializer.

    The ``func`` argument must take two arguments (i.e. the dict containing the
    serialized values and the type that the values should be deserialized into)
    and also a ``kwargs`` parameter. For example:

    >>> def func(dict_, cls, **kwargs):
    ...    return cls()

    You may ask additional arguments between ``cls`` and ``kwargs``.

    :param func: the deserializer function.
    :param cls: the type or the name of the type this serializer can handle.
    :param high_prio: determines the order in which is looked for the callable.
    :param fork_inst: if given, it uses this fork of ``JsonSerializable``.
    :return: None.
    """
    if cls:
        index = 0 if high_prio else len(fork_inst._classes_deserializers)
        fork_inst._classes_deserializers.insert(index, cls)
        cls_name = get_class_name(cls, fork_inst=fork_inst)
        fork_inst._deserializers[cls_name.lower()] = func
    else:
        fork_inst._deserializers['nonetype'] = func


def suppress_warnings(
        do_suppress: Optional[bool] = True,
        fork_inst: Optional[type] = StateHolder):
    """
    Suppress (or stop suppressing) warnings.
    :param do_suppress: if ``True``, warnings will be suppressed from now on.
    :param fork_inst: if given, it uses this fork of ``JsonSerializable``.
    :return: None.
    """
    fork_inst._suppress_warnings = do_suppress


def announce_class(
        cls: type,
        cls_name: Optional[str] = None,
        fork_inst: type = StateHolder):
    """
    Announce the given cls to jsons to allow jsons to deserialize a verbose
    dump into that class.
    :param cls: the class that is to be announced.
    :param cls_name: a custom name for that class.
    :param fork_inst: if given, it uses this fork of ``JsonSerializable``.
    :return: None.
    """
    cls_name = cls_name or get_class_name(cls, fully_qualified=True,
                                          fork_inst=fork_inst)
    fork_inst._announced_classes[cls] = cls_name
    fork_inst._announced_classes[cls_name] = cls


def _check_and_get_cls(json_obj: object, cls: type, fork_inst: type) -> type:
    # Check if json_obj is of a valid type and return the cls.
    if type(json_obj) not in VALID_TYPES:
        invalid_type = get_class_name(type(json_obj), fork_inst=fork_inst)
        valid_types = [get_class_name(typ, fork_inst=fork_inst)
                       for typ in VALID_TYPES]
        msg = ('Invalid type: "{}", only arguments of the following types are '
               'allowed: {}'.format(invalid_type, ", ".join(valid_types)))
        raise DeserializationError(msg, json_obj, cls)
    if json_obj is None:
        raise DeserializationError('Cannot load None with strict=True',
                                   json_obj, cls)

    cls_from_meta, meta = _get_cls_and_meta(json_obj, fork_inst)
    return cls or cls_from_meta or type(json_obj)


def _get_cls_and_meta(
        json_obj: object,
        fork_inst: type) -> Tuple[Optional[type], Optional[dict]]:
    if isinstance(json_obj, dict) and META_ATTR in json_obj:
        cls_str = json_obj[META_ATTR]['classes']['/']
        cls = _get_cls_from_str(cls_str, json_obj, fork_inst)
        return cls, json_obj[META_ATTR]
    return None, None


def _get_cls_from_str(cls_str: str, source: object, fork_inst) -> type:
    try:
        # importlib.import_module('jsons.exceptions')
        splitted = cls_str.split('.')
        module_name = '.'.join(splitted[:-1])
        cls_name = splitted[-1]
        cls_module = import_module(module_name)
        cls = getattr(cls_module, cls_name)
        if not cls or not isinstance(cls, type):
            cls = _lookup_announced_class(cls_str, source, fork_inst)
    except (ImportError, AttributeError, ValueError):
        cls = _lookup_announced_class(cls_str, source, fork_inst)
    return cls


def _lookup_announced_class(
        cls_str: str,
        source: object,
        fork_inst: type) -> type:
    cls = fork_inst._announced_classes.get(cls_str)
    if not cls:
        msg = ('Could not find a suitable type for "{}". Make sure it can be '
               'imported or that is has been announced.'.format(cls_str))
        raise UnknownClassError(msg, source, cls_str)
    return cls


1			"""
2			PRIVATE MODULE: do not import (from) it directly.
3
4			This module contains the implementation of the main functions of jsons, such
5			as `load` and `dump`.
6			"""
7			import json
8			from importlib import import_module
9			from json import JSONDecodeError
10			from typing import Dict, Callable, Optional, Union, Tuple
11			from jsons._common_impl import (
12			get_class_name,
13			get_parents,
14			META_ATTR,
15			StateHolder
16			)
17			from jsons.exceptions import (
18			DecodeError,
19			DeserializationError,
20			JsonsError,
21			SerializationError,
22			UnknownClassError
23			)
24
25			VALID_TYPES = (str, int, float, bool, list, tuple, set, dict, type(None))
26			RFC3339_DATETIME_PATTERN = '%Y-%m-%dT%H:%M:%S'
27
28
29			def dump(obj: object,
30			cls: Optional[type] = None,
31			fork_inst: Optional[type] = StateHolder,
32			**kwargs) -> object:
33			"""
34			Serialize the given ``obj`` to a JSON equivalent type (e.g. dict, list,
35			int, ...).
36
37			The way objects are serialized can be finetuned by setting serializer
38			functions for the specific type using ``set_serializer``.
39
40			You can also provide ``cls`` to specify that ``obj`` needs to be serialized
41			as if it was of type ``cls`` (meaning to only take into account attributes
42			from ``cls``). The type ``cls`` must have a ``__slots__`` defined. Any type
43			will do, but in most cases you may want ``cls`` to be a base class of
44			``obj``.
45			:param obj: a Python instance of any sort.
46			:param cls: if given, ``obj`` will be dumped as if it is of type ``type``.
47			:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
48			:param kwargs: the keyword args are passed on to the serializer function.
49			:return: the serialized obj as a JSON type.
50			"""
51			if cls and not hasattr(cls, '__slots__'):
52			raise SerializationError('Invalid type: "{}". Only types that have a '
53			'__slots__ defined are allowed when '
54			'providing "cls".'
55			.format(get_class_name(cls, fork_inst=fork_inst)))
56			cls_ = cls or obj.__class__
57			serializer = _get_serializer(cls_, fork_inst)
58			kwargs_ = {
59			'fork_inst': fork_inst,
60			**kwargs
61			}
62			try:
63			return serializer(obj, cls=cls, **kwargs_)
64			except Exception as err:
65			raise SerializationError(str(err))
66
67
68			def load(json_obj: object,
69			cls: Optional[type] = None,
70			strict: bool = False,
71			fork_inst: Optional[type] = StateHolder,
72			attr_getters: Optional[Dict[str, Callable[[], object]]] = None,
73			**kwargs) -> object:
74			"""
75			Deserialize the given ``json_obj`` to an object of type ``cls``. If the
76			contents of ``json_obj`` do not match the interface of ``cls``, a
77			DeserializationError is raised.
78
79			If ``json_obj`` contains a value that belongs to a custom class, there must
80			be a type hint present for that value in ``cls`` to let this function know
81			what type it should deserialize that value to.
82
83
84			Example:
85
86			>>> from typing import List
87			>>> import jsons
88			>>> class Person:
89			... # No type hint required for name
90			... def __init__(self, name):
91			... self.name = name
92			>>> class Family:
93			... # Person is a custom class, use a type hint
94			... def __init__(self, persons: List[Person]):
95			... self.persons = persons
96			>>> loaded = jsons.load({'persons': [{'name': 'John'}]}, Family)
97			>>> loaded.persons[0].name
98			'John'
99
100			If no ``cls`` is given, a dict is simply returned, but contained values
101			(e.g. serialized ``datetime`` values) are still deserialized.
102
103			If `strict` mode is off and the type of `json_obj` exactly matches `cls`
104			then `json_obj` is simply returned.
105
106			:param json_obj: the dict that is to be deserialized.
107			:param cls: a matching class of which an instance should be returned.
108			:param strict: a bool to determine if the deserializer should be strict
109			(i.e. fail on a partially deserialized `json_obj` or on `None`).
110			:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
111			:param attr_getters: a ``dict`` that may hold callables that return values
112			for certain attributes.
113			:param kwargs: the keyword args are passed on to the deserializer function.
114			:return: an instance of ``cls`` if given, a dict otherwise.
115			"""
116			if not strict and (json_obj is None or type(json_obj) == cls):
117			return json_obj
118			cls = _check_and_get_cls(json_obj, cls, fork_inst)
119			deserializer = _get_deserializer(cls, fork_inst)
120			kwargs_ = {
121			'strict': strict,
122			'fork_inst': fork_inst,
123			'attr_getters': attr_getters,
124			**kwargs
125			}
126			try:
127			return deserializer(json_obj, cls, **kwargs_)
128			except Exception as err:
129			if isinstance(err, JsonsError):
130			raise
131			raise DeserializationError(str(err), json_obj, cls)
132
133
134			def _get_serializer(cls: type,
135			fork_inst: Optional[type] = StateHolder) -> callable:
136			serializer = _get_lizer(cls, fork_inst._serializers,
137			fork_inst._classes_serializers, fork_inst)
138			return serializer
139
140
141			def _get_deserializer(cls: type,
142			fork_inst: Optional[type] = StateHolder) -> callable:
143			deserializer = _get_lizer(cls, fork_inst._deserializers,
144			fork_inst._classes_deserializers, fork_inst)
145			return deserializer
146
147
148			def _get_lizer(cls: type,
149			lizers: Dict[str, callable],
150			classes_lizers: list,
151			fork_inst: type) -> callable:
152			cls_name = get_class_name(cls, str.lower, fork_inst=fork_inst)
153			lizer = lizers.get(cls_name, None)
154			if not lizer:
155			parents = get_parents(cls, classes_lizers)
156			if parents:
157			pname = get_class_name(parents[0], str.lower, fork_inst=fork_inst)
158			lizer = lizers[pname]
159			return lizer
160
161
162			def dumps(obj: object,
163			jdkwargs: Optional[Dict[str, object]] = None,
164			*args,
165			**kwargs) -> str:
166			"""
167			Extend ``json.dumps``, allowing any Python instance to be dumped to a
168			string. Any extra (keyword) arguments are passed on to ``json.dumps``.
169
170			:param obj: the object that is to be dumped to a string.
171			:param jdkwargs: extra keyword arguments for ``json.dumps`` (not
172			``jsons.dumps``!)
173			:param args: extra arguments for ``jsons.dumps``.
174			:param kwargs: Keyword arguments that are passed on through the
175			serialization process.
176			passed on to the serializer function.
177			:return: ``obj`` as a ``str``.
178			"""
179			jdkwargs = jdkwargs or {}
180			dumped = dump(obj, args, *kwargs)
181			return json.dumps(dumped, **jdkwargs)
182
183
184			def loads(str_: str,
185			cls: Optional[type] = None,
186			jdkwargs: Optional[Dict[str, object]] = None,
187			*args,
188			**kwargs) -> object:
189			"""
190			Extend ``json.loads``, allowing a string to be loaded into a dict or a
191			Python instance of type ``cls``. Any extra (keyword) arguments are passed
192			on to ``json.loads``.
193
194			:param str_: the string that is to be loaded.
195			:param cls: a matching class of which an instance should be returned.
196			:param jdkwargs: extra keyword arguments for ``json.loads`` (not
197			``jsons.loads``!)
198			:param args: extra arguments for ``jsons.loads``.
199			:param kwargs: extra keyword arguments for ``jsons.loads``.
200			:return: a JSON-type object (dict, str, list, etc.) or an instance of type
201			``cls`` if given.
202			"""
203			jdkwargs = jdkwargs or {}
204			try:
205			obj = json.loads(str_, **jdkwargs)
206			except JSONDecodeError as err:
207			raise DecodeError('Could not load a dict; the given string is not '
208			'valid JSON.', str_, cls, err)
209			else:
210			return load(obj, cls, args, *kwargs)
211
212
213			def dumpb(obj: object,
214			encoding: str = 'utf-8',
215			jdkwargs: Optional[Dict[str, object]] = None,
216			*args,
217			**kwargs) -> bytes:
218			"""
219			Extend ``json.dumps``, allowing any Python instance to be dumped to bytes.
220			Any extra (keyword) arguments are passed on to ``json.dumps``.
221
222			:param obj: the object that is to be dumped to bytes.
223			:param encoding: the encoding that is used to transform to bytes.
224			:param jdkwargs: extra keyword arguments for ``json.dumps`` (not
225			``jsons.dumps``!)
226			:param args: extra arguments for ``jsons.dumps``.
227			:param kwargs: Keyword arguments that are passed on through the
228			serialization process.
229			passed on to the serializer function.
230			:return: ``obj`` as ``bytes``.
231			"""
232			jdkwargs = jdkwargs or {}
233			dumped_dict = dump(obj, args, *kwargs)
234			dumped_str = json.dumps(dumped_dict, **jdkwargs)
235			return dumped_str.encode(encoding=encoding)
236
237
238			def loadb(bytes_: bytes,
239			cls: Optional[type] = None,
240			encoding: str = 'utf-8',
241			jdkwargs: Optional[Dict[str, object]] = None,
242			*args,
243			**kwargs) -> object:
244			"""
245			Extend ``json.loads``, allowing bytes to be loaded into a dict or a Python
246			instance of type ``cls``. Any extra (keyword) arguments are passed on to
247			``json.loads``.
248
249			:param bytes_: the bytes that are to be loaded.
250			:param cls: a matching class of which an instance should be returned.
251			:param encoding: the encoding that is used to transform from bytes.
252			:param jdkwargs: extra keyword arguments for ``json.loads`` (not
253			``jsons.loads``!)
254			:param args: extra arguments for ``jsons.loads``.
255			:param kwargs: extra keyword arguments for ``jsons.loads``.
256			:return: a JSON-type object (dict, str, list, etc.) or an instance of type
257			``cls`` if given.
258			"""
259			if not isinstance(bytes_, bytes):
260			raise DeserializationError('loadb accepts bytes only, "{}" was given'
261			.format(type(bytes_)), bytes_, cls)
262			jdkwargs = jdkwargs or {}
263			str_ = bytes_.decode(encoding=encoding)
264			return loads(str_, cls, jdkwargs=jdkwargs, args, *kwargs)
265
266
267			def set_serializer(func: callable,
268			cls: type,
269			high_prio: bool = True,
270			fork_inst: type = StateHolder) -> None:
271			"""
272			Set a serializer function for the given type. You may override the default
273			behavior of ``jsons.load`` by setting a custom serializer.
274
275			The ``func`` argument must take one argument (i.e. the object that is to be
276			serialized) and also a ``kwargs`` parameter. For example:
277
278			>>> def func(obj, **kwargs):
279			... return dict()
280
281			You may ask additional arguments between ``cls`` and ``kwargs``.
282
283			:param func: the serializer function.
284			:param cls: the type this serializer can handle.
285			:param high_prio: determines the order in which is looked for the callable.
286			:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
287			:return: None.
288			"""
289			if cls:
290			index = 0 if high_prio else len(fork_inst._classes_serializers)
291			fork_inst._classes_serializers.insert(index, cls)
292			cls_name = get_class_name(cls, fork_inst=fork_inst)
293			fork_inst._serializers[cls_name.lower()] = func
294			else:
295			fork_inst._serializers['nonetype'] = func
296
297
298			def set_deserializer(func: callable,
299			cls: Union[type, str],
300			high_prio: bool = True,
301			fork_inst: type = StateHolder) -> None:
302			"""
303			Set a deserializer function for the given type. You may override the
304			default behavior of ``jsons.dump`` by setting a custom deserializer.
305
306			The ``func`` argument must take two arguments (i.e. the dict containing the
307			serialized values and the type that the values should be deserialized into)
308			and also a ``kwargs`` parameter. For example:
309
310			>>> def func(dict_, cls, **kwargs):
311			... return cls()
312
313			You may ask additional arguments between ``cls`` and ``kwargs``.
314
315			:param func: the deserializer function.
316			:param cls: the type or the name of the type this serializer can handle.
317			:param high_prio: determines the order in which is looked for the callable.
318			:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
319			:return: None.
320			"""
321			if cls:
322			index = 0 if high_prio else len(fork_inst._classes_deserializers)
323			fork_inst._classes_deserializers.insert(index, cls)
324			cls_name = get_class_name(cls, fork_inst=fork_inst)
325			fork_inst._deserializers[cls_name.lower()] = func
326			else:
327			fork_inst._deserializers['nonetype'] = func
328
329
330			def suppress_warnings(
331			do_suppress: Optional[bool] = True,
332			fork_inst: Optional[type] = StateHolder):
333			"""
334			Suppress (or stop suppressing) warnings.
335			:param do_suppress: if ``True``, warnings will be suppressed from now on.
336			:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
337			:return: None.
338			"""
339			fork_inst._suppress_warnings = do_suppress
340
341
342			def announce_class(
343			cls: type,
344			cls_name: Optional[str] = None,
345			fork_inst: type = StateHolder):
346			"""
347			Announce the given cls to jsons to allow jsons to deserialize a verbose
348			dump into that class.
349			:param cls: the class that is to be announced.
350			:param cls_name: a custom name for that class.
351			:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
352			:return: None.
353			"""
354			cls_name = cls_name or get_class_name(cls, fully_qualified=True,
355			fork_inst=fork_inst)
356			fork_inst._announced_classes[cls] = cls_name
357			fork_inst._announced_classes[cls_name] = cls
358
359
360			def _check_and_get_cls(json_obj: object, cls: type, fork_inst: type) -> type:
361			# Check if json_obj is of a valid type and return the cls.
362			if type(json_obj) not in VALID_TYPES:
363			invalid_type = get_class_name(type(json_obj), fork_inst=fork_inst)
364			valid_types = [get_class_name(typ, fork_inst=fork_inst)
365			for typ in VALID_TYPES]
366			msg = ('Invalid type: "{}", only arguments of the following types are '
367			'allowed: {}'.format(invalid_type, ", ".join(valid_types)))
368			raise DeserializationError(msg, json_obj, cls)
369			if json_obj is None:
370			raise DeserializationError('Cannot load None with strict=True',
371			json_obj, cls)
372
373			cls_from_meta, meta = _get_cls_and_meta(json_obj, fork_inst)
374			return cls or cls_from_meta or type(json_obj)
375
376
377			def _get_cls_and_meta(
378			json_obj: object,
379			fork_inst: type) -> Tuple[Optional[type], Optional[dict]]:
380			if isinstance(json_obj, dict) and META_ATTR in json_obj:
381			cls_str = json_obj[META_ATTR]['classes']['/']
382			cls = _get_cls_from_str(cls_str, json_obj, fork_inst)
383			return cls, json_obj[META_ATTR]
384			return None, None
385
386
387			def _get_cls_from_str(cls_str: str, source: object, fork_inst) -> type:
388			try:
389			# importlib.import_module('jsons.exceptions')
390			splitted = cls_str.split('.')
391			module_name = '.'.join(splitted[:-1])
392			cls_name = splitted[-1]
393			cls_module = import_module(module_name)
394			cls = getattr(cls_module, cls_name)
395			if not cls or not isinstance(cls, type):
396			cls = _lookup_announced_class(cls_str, source, fork_inst)
397			except (ImportError, AttributeError, ValueError):
398			cls = _lookup_announced_class(cls_str, source, fork_inst)
399			return cls
400
401
402			def _lookup_announced_class(
403			cls_str: str,
404			source: object,
405			fork_inst: type) -> type:
406			cls = fork_inst._announced_classes.get(cls_str)
407			if not cls:
408			msg = ('Could not find a suitable type for "{}". Make sure it can be '
409			'imported or that is has been announced.'.format(cls_str))
410			raise UnknownClassError(msg, source, cls_str)
411			return cls
412

ramonhagenaars / jsons

Pull Request — master (#23)

jsons._main_impl.set_deserializer() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like