jsons._main_impl.loadb() - Code Metrics - Inspection of "Version bump." - ramonhagenaars/jsons - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 988c2c...4e149e )

by Ramon

created 2019-03-20 19:52 UTC

jsons._main_impl.loadb() A

↳ Parent: jsons._main_impl

Complexity

Conditions

Size

Total Lines	27
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	12
nop	6
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 json import JSONDecodeError
from typing import Dict, Callable, Optional, Union, Tuple
from jsons._common_impl import (
    get_class_name,
    get_parents,
    META_ATTR,
    StateHolder,
    get_cls_from_str)
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, meta_hints = _check_and_get_cls_and_meta_hints(json_obj, cls,
                                                        fork_inst)
    deserializer = _get_deserializer(cls, fork_inst)
    kwargs_ = {
        'strict': strict,
        'fork_inst': fork_inst,
        'attr_getters': attr_getters,
        'meta_hints': meta_hints,
        **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_and_meta_hints(
        json_obj: object,
        cls: type,
        fork_inst: type) -> Tuple[type, Optional[dict]]:
    # 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)
    meta_hints = meta.get('classes', {}) if meta else {}
    # Providing cls takes precedence over cls_from_meta.
    return (cls or cls_from_meta or type(json_obj)), meta_hints


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 _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 json import JSONDecodeError
9			from typing import Dict, Callable, Optional, Union, Tuple
10			from jsons._common_impl import (
11			get_class_name,
12			get_parents,
13			META_ATTR,
14			StateHolder,
15			get_cls_from_str)
16			from jsons.exceptions import (
17			DecodeError,
18			DeserializationError,
19			JsonsError,
20			SerializationError,
21			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, meta_hints = _check_and_get_cls_and_meta_hints(json_obj, cls,
118			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			'meta_hints': meta_hints,
125			**kwargs
126			}
127			try:
128			return deserializer(json_obj, cls, **kwargs_)
129			except Exception as err:
130			if isinstance(err, JsonsError):
131			raise
132			raise DeserializationError(str(err), json_obj, cls)
133
134
135			def _get_serializer(cls: type,
136			fork_inst: Optional[type] = StateHolder) -> callable:
137			serializer = _get_lizer(cls, fork_inst._serializers,
138			fork_inst._classes_serializers, fork_inst)
139			return serializer
140
141
142			def _get_deserializer(cls: type,
143			fork_inst: Optional[type] = StateHolder) -> callable:
144			deserializer = _get_lizer(cls, fork_inst._deserializers,
145			fork_inst._classes_deserializers, fork_inst)
146			return deserializer
147
148
149			def _get_lizer(cls: type,
150			lizers: Dict[str, callable],
151			classes_lizers: list,
152			fork_inst: type) -> callable:
153			cls_name = get_class_name(cls, str.lower, fork_inst=fork_inst)
154			lizer = lizers.get(cls_name, None)
155			if not lizer:
156			parents = get_parents(cls, classes_lizers)
157			if parents:
158			pname = get_class_name(parents[0], str.lower, fork_inst=fork_inst)
159			lizer = lizers[pname]
160			return lizer
161
162
163			def dumps(obj: object,
164			jdkwargs: Optional[Dict[str, object]] = None,
165			*args,
166			**kwargs) -> str:
167			"""
168			Extend ``json.dumps``, allowing any Python instance to be dumped to a
169			string. Any extra (keyword) arguments are passed on to ``json.dumps``.
170
171			:param obj: the object that is to be dumped to a string.
172			:param jdkwargs: extra keyword arguments for ``json.dumps`` (not
173			``jsons.dumps``!)
174			:param args: extra arguments for ``jsons.dumps``.
175			:param kwargs: Keyword arguments that are passed on through the
176			serialization process.
177			passed on to the serializer function.
178			:return: ``obj`` as a ``str``.
179			"""
180			jdkwargs = jdkwargs or {}
181			dumped = dump(obj, args, *kwargs)
182			return json.dumps(dumped, **jdkwargs)
183
184
185			def loads(str_: str,
186			cls: Optional[type] = None,
187			jdkwargs: Optional[Dict[str, object]] = None,
188			*args,
189			**kwargs) -> object:
190			"""
191			Extend ``json.loads``, allowing a string to be loaded into a dict or a
192			Python instance of type ``cls``. Any extra (keyword) arguments are passed
193			on to ``json.loads``.
194
195			:param str_: the string that is to be loaded.
196			:param cls: a matching class of which an instance should be returned.
197			:param jdkwargs: extra keyword arguments for ``json.loads`` (not
198			``jsons.loads``!)
199			:param args: extra arguments for ``jsons.loads``.
200			:param kwargs: extra keyword arguments for ``jsons.loads``.
201			:return: a JSON-type object (dict, str, list, etc.) or an instance of type
202			``cls`` if given.
203			"""
204			jdkwargs = jdkwargs or {}
205			try:
206			obj = json.loads(str_, **jdkwargs)
207			except JSONDecodeError as err:
208			raise DecodeError('Could not load a dict; the given string is not '
209			'valid JSON.', str_, cls, err)
210			else:
211			return load(obj, cls, args, *kwargs)
212
213
214			def dumpb(obj: object,
215			encoding: str = 'utf-8',
216			jdkwargs: Optional[Dict[str, object]] = None,
217			*args,
218			**kwargs) -> bytes:
219			"""
220			Extend ``json.dumps``, allowing any Python instance to be dumped to bytes.
221			Any extra (keyword) arguments are passed on to ``json.dumps``.
222
223			:param obj: the object that is to be dumped to bytes.
224			:param encoding: the encoding that is used to transform to bytes.
225			:param jdkwargs: extra keyword arguments for ``json.dumps`` (not
226			``jsons.dumps``!)
227			:param args: extra arguments for ``jsons.dumps``.
228			:param kwargs: Keyword arguments that are passed on through the
229			serialization process.
230			passed on to the serializer function.
231			:return: ``obj`` as ``bytes``.
232			"""
233			jdkwargs = jdkwargs or {}
234			dumped_dict = dump(obj, args, *kwargs)
235			dumped_str = json.dumps(dumped_dict, **jdkwargs)
236			return dumped_str.encode(encoding=encoding)
237
238
239			def loadb(bytes_: bytes,
240			cls: Optional[type] = None,
241			encoding: str = 'utf-8',
242			jdkwargs: Optional[Dict[str, object]] = None,
243			*args,
244			**kwargs) -> object:
245			"""
246			Extend ``json.loads``, allowing bytes to be loaded into a dict or a Python
247			instance of type ``cls``. Any extra (keyword) arguments are passed on to
248			``json.loads``.
249
250			:param bytes_: the bytes that are to be loaded.
251			:param cls: a matching class of which an instance should be returned.
252			:param encoding: the encoding that is used to transform from bytes.
253			:param jdkwargs: extra keyword arguments for ``json.loads`` (not
254			``jsons.loads``!)
255			:param args: extra arguments for ``jsons.loads``.
256			:param kwargs: extra keyword arguments for ``jsons.loads``.
257			:return: a JSON-type object (dict, str, list, etc.) or an instance of type
258			``cls`` if given.
259			"""
260			if not isinstance(bytes_, bytes):
261			raise DeserializationError('loadb accepts bytes only, "{}" was given'
262			.format(type(bytes_)), bytes_, cls)
263			jdkwargs = jdkwargs or {}
264			str_ = bytes_.decode(encoding=encoding)
265			return loads(str_, cls, jdkwargs=jdkwargs, args, *kwargs)
266
267
268			def set_serializer(func: callable,
269			cls: type,
270			high_prio: bool = True,
271			fork_inst: type = StateHolder) -> None:
272			"""
273			Set a serializer function for the given type. You may override the default
274			behavior of ``jsons.load`` by setting a custom serializer.
275
276			The ``func`` argument must take one argument (i.e. the object that is to be
277			serialized) and also a ``kwargs`` parameter. For example:
278
279			>>> def func(obj, **kwargs):
280			... return dict()
281
282			You may ask additional arguments between ``cls`` and ``kwargs``.
283
284			:param func: the serializer function.
285			:param cls: the type this serializer can handle.
286			:param high_prio: determines the order in which is looked for the callable.
287			:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
288			:return: None.
289			"""
290			if cls:
291			index = 0 if high_prio else len(fork_inst._classes_serializers)
292			fork_inst._classes_serializers.insert(index, cls)
293			cls_name = get_class_name(cls, fork_inst=fork_inst)
294			fork_inst._serializers[cls_name.lower()] = func
295			else:
296			fork_inst._serializers['nonetype'] = func
297
298
299			def set_deserializer(func: callable,
300			cls: Union[type, str],
301			high_prio: bool = True,
302			fork_inst: type = StateHolder) -> None:
303			"""
304			Set a deserializer function for the given type. You may override the
305			default behavior of ``jsons.dump`` by setting a custom deserializer.
306
307			The ``func`` argument must take two arguments (i.e. the dict containing the
308			serialized values and the type that the values should be deserialized into)
309			and also a ``kwargs`` parameter. For example:
310
311			>>> def func(dict_, cls, **kwargs):
312			... return cls()
313
314			You may ask additional arguments between ``cls`` and ``kwargs``.
315
316			:param func: the deserializer function.
317			:param cls: the type or the name of the type this serializer can handle.
318			:param high_prio: determines the order in which is looked for the callable.
319			:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
320			:return: None.
321			"""
322			if cls:
323			index = 0 if high_prio else len(fork_inst._classes_deserializers)
324			fork_inst._classes_deserializers.insert(index, cls)
325			cls_name = get_class_name(cls, fork_inst=fork_inst)
326			fork_inst._deserializers[cls_name.lower()] = func
327			else:
328			fork_inst._deserializers['nonetype'] = func
329
330
331			def suppress_warnings(
332			do_suppress: Optional[bool] = True,
333			fork_inst: Optional[type] = StateHolder):
334			"""
335			Suppress (or stop suppressing) warnings.
336			:param do_suppress: if ``True``, warnings will be suppressed from now on.
337			:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
338			:return: None.
339			"""
340			fork_inst._suppress_warnings = do_suppress
341
342
343			def announce_class(
344			cls: type,
345			cls_name: Optional[str] = None,
346			fork_inst: type = StateHolder):
347			"""
348			Announce the given cls to jsons to allow jsons to deserialize a verbose
349			dump into that class.
350			:param cls: the class that is to be announced.
351			:param cls_name: a custom name for that class.
352			:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
353			:return: None.
354			"""
355			cls_name = cls_name or get_class_name(cls, fully_qualified=True,
356			fork_inst=fork_inst)
357			fork_inst._announced_classes[cls] = cls_name
358			fork_inst._announced_classes[cls_name] = cls
359
360
361			def _check_and_get_cls_and_meta_hints(
362			json_obj: object,
363			cls: type,
364			fork_inst: type) -> Tuple[type, Optional[dict]]:
365			# Check if json_obj is of a valid type and return the cls.
366			if type(json_obj) not in VALID_TYPES:
367			invalid_type = get_class_name(type(json_obj), fork_inst=fork_inst)
368			valid_types = [get_class_name(typ, fork_inst=fork_inst)
369			for typ in VALID_TYPES]
370			msg = ('Invalid type: "{}", only arguments of the following types are '
371			'allowed: {}'.format(invalid_type, ", ".join(valid_types)))
372			raise DeserializationError(msg, json_obj, cls)
373			if json_obj is None:
374			raise DeserializationError('Cannot load None with strict=True',
375			json_obj, cls)
376
377			cls_from_meta, meta = _get_cls_and_meta(json_obj, fork_inst)
378			meta_hints = meta.get('classes', {}) if meta else {}
379			# Providing cls takes precedence over cls_from_meta.
380			return (cls or cls_from_meta or type(json_obj)), meta_hints
381
382
383			def _get_cls_and_meta(
384			json_obj: object,
385			fork_inst: type) -> Tuple[Optional[type], Optional[dict]]:
386			if isinstance(json_obj, dict) and META_ATTR in json_obj:
387			cls_str = json_obj[META_ATTR]['classes']['/']
388			cls = get_cls_from_str(cls_str, json_obj, fork_inst)
389			return cls, json_obj[META_ATTR]
390			return None, None
391
392
393			def _lookup_announced_class(
394			cls_str: str,
395			source: object,
396			fork_inst: type) -> type:
397			cls = fork_inst._announced_classes.get(cls_str)
398			if not cls:
399			msg = ('Could not find a suitable type for "{}". Make sure it can be '
400			'imported or that is has been announced.'.format(cls_str))
401			raise UnknownClassError(msg, source, cls_str)
402			return cls
403

ramonhagenaars / jsons

Push — master ( 988c2c...4e149e )

jsons._main_impl.loadb() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like