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

Completed

Pull Request — master (#23)

by Ramon

created 2019-03-15 20:37 UTC

jsons._main_impl.loads() A

↳ Parent: jsons._main_impl

Complexity

Conditions

Size

Total Lines	27
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

ramonhagenaars / jsons

Pull Request — master (#23)

jsons._main_impl.loads() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like