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

Passed

Pull Request — master (#42)

by Ramon

created 2019-04-07 20:33 UTC

jsons._main_impl._should_skip() A

↳ Parent: jsons._main_impl

Complexity

Conditions

Size

Total Lines	3
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

"""
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, Sequence
from jsons._common_impl import (
    get_class_name,
    get_parents,
    StateHolder,
    get_cls_from_str, get_cls_and_meta, determine_precedence
)
from jsons.exceptions import (
    DecodeError,
    DeserializationError,
    JsonsError,
    SerializationError
)

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,
                                                fully_qualified=True)))
    cls_ = cls or obj.__class__
    serializer = _get_serializer(cls_, fork_inst)
    kwargs_ = {
        'fork_inst': fork_inst,
        **kwargs
    }
    announce_class(cls_, fork_inst=fork_inst)
    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 _should_skip(json_obj, cls, strict):
        return json_obj
    if isinstance(cls, str):
        cls = get_cls_from_str(cls, json_obj, fork_inst)
    cls, meta_hints = _check_and_get_cls_and_meta_hints(
        json_obj, cls, fork_inst, kwargs.get('_inferred_cls', False))

    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 _should_skip(json_obj: object, cls: type, strict: bool):
    if not strict and (json_obj is None or type(json_obj) == cls):
        return True


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,
               recursive: bool = False) -> callable:
    cls_name = get_class_name(cls, str.lower, fork_inst=fork_inst,
                              fully_qualified=True)
    lizer = (lizers.get(cls_name, None)
             or _get_lizer_by_parents(cls, lizers, classes_lizers, fork_inst))
    if not lizer and not recursive:
        return _get_lizer(cls.__supertype__, lizers,
                          classes_lizers, fork_inst, True)
    return lizer


def _get_lizer_by_parents(cls: type,
                          lizers: Dict[str, callable],
                          classes_lizers: list,
                          fork_inst: type) -> callable:
    result = None
    parents = get_parents(cls, classes_lizers)
    if parents:
        pname = get_class_name(parents[0], str.lower, fork_inst=fork_inst,
                               fully_qualified=True)
        result = lizers[pname]
    return result


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: Union[type, Sequence[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 or sequence of types 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 isinstance(cls, Sequence):
        for cls_ in cls:
            set_serializer(func, cls_, high_prio, fork_inst)
    elif 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,
                                  fully_qualified=True)
        fork_inst._serializers[cls_name.lower()] = func
    else:
        fork_inst._serializers['nonetype'] = func


def set_deserializer(func: callable,
                     cls: Union[type, Sequence[type]],
                     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 sequence of types 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 isinstance(cls, Sequence):
        for cls_ in cls:
            set_deserializer(func, cls_, high_prio, fork_inst)
    elif 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,
                                  fully_qualified=True)
        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,
        inferred_cls: bool) -> 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,
                                      fully_qualified=True)
        valid_types = [get_class_name(typ, fork_inst=fork_inst,
                                      fully_qualified=True)
                       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 {}
    return determine_precedence(
        cls, cls_from_meta, type(json_obj), inferred_cls), meta_hints


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, Sequence
10			from jsons._common_impl import (
11			get_class_name,
12			get_parents,
13			StateHolder,
14			get_cls_from_str, get_cls_and_meta, determine_precedence
15			)
16			from jsons.exceptions import (
17			DecodeError,
18			DeserializationError,
19			JsonsError,
20			SerializationError
21			)
22
23			VALID_TYPES = (str, int, float, bool, list, tuple, set, dict, type(None))
24			RFC3339_DATETIME_PATTERN = '%Y-%m-%dT%H:%M:%S'
25
26
27			def dump(obj: object,
28			cls: Optional[type] = None,
29			fork_inst: Optional[type] = StateHolder,
30			**kwargs) -> object:
31			"""
32			Serialize the given ``obj`` to a JSON equivalent type (e.g. dict, list,
33			int, ...).
34
35			The way objects are serialized can be finetuned by setting serializer
36			functions for the specific type using ``set_serializer``.
37
38			You can also provide ``cls`` to specify that ``obj`` needs to be serialized
39			as if it was of type ``cls`` (meaning to only take into account attributes
40			from ``cls``). The type ``cls`` must have a ``__slots__`` defined. Any type
41			will do, but in most cases you may want ``cls`` to be a base class of
42			``obj``.
43			:param obj: a Python instance of any sort.
44			:param cls: if given, ``obj`` will be dumped as if it is of type ``type``.
45			:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
46			:param kwargs: the keyword args are passed on to the serializer function.
47			:return: the serialized obj as a JSON type.
48			"""
49			if cls and not hasattr(cls, '__slots__'):
50			raise SerializationError('Invalid type: "{}". Only types that have a '
51			'__slots__ defined are allowed when '
52			'providing "cls".'
53			.format(get_class_name(cls, fork_inst=fork_inst,
54			fully_qualified=True)))
55			cls_ = cls or obj.__class__
56			serializer = _get_serializer(cls_, fork_inst)
57			kwargs_ = {
58			'fork_inst': fork_inst,
59			**kwargs
60			}
61			announce_class(cls_, fork_inst=fork_inst)
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 _should_skip(json_obj, cls, strict):
117			return json_obj
118			if isinstance(cls, str):
119			cls = get_cls_from_str(cls, json_obj, fork_inst)
120			cls, meta_hints = _check_and_get_cls_and_meta_hints(
121			json_obj, cls, fork_inst, kwargs.get('_inferred_cls', False))
122
123			deserializer = _get_deserializer(cls, fork_inst)
124			kwargs_ = {
125			'strict': strict,
126			'fork_inst': fork_inst,
127			'attr_getters': attr_getters,
128			'meta_hints': meta_hints,
129			**kwargs
130			}
131			try:
132			return deserializer(json_obj, cls, **kwargs_)
133			except Exception as err:
134			if isinstance(err, JsonsError):
135			raise
136			raise DeserializationError(str(err), json_obj, cls)
137
138
139			def _should_skip(json_obj: object, cls: type, strict: bool):
140			if not strict and (json_obj is None or type(json_obj) == cls):
141			return True
142
143
144			def _get_serializer(cls: type,
145			fork_inst: Optional[type] = StateHolder) -> callable:
146			serializer = _get_lizer(cls, fork_inst._serializers,
147			fork_inst._classes_serializers, fork_inst)
148			return serializer
149
150
151			def _get_deserializer(cls: type,
152			fork_inst: Optional[type] = StateHolder) -> callable:
153			deserializer = _get_lizer(cls, fork_inst._deserializers,
154			fork_inst._classes_deserializers, fork_inst)
155			return deserializer
156
157
158			def _get_lizer(cls: type,
159			lizers: Dict[str, callable],
160			classes_lizers: list,
161			fork_inst: type,
162			recursive: bool = False) -> callable:
163			cls_name = get_class_name(cls, str.lower, fork_inst=fork_inst,
164			fully_qualified=True)
165			lizer = (lizers.get(cls_name, None)
166			or _get_lizer_by_parents(cls, lizers, classes_lizers, fork_inst))
167			if not lizer and not recursive:
168			return _get_lizer(cls.__supertype__, lizers,
169			classes_lizers, fork_inst, True)
170			return lizer
171
172
173			def _get_lizer_by_parents(cls: type,
174			lizers: Dict[str, callable],
175			classes_lizers: list,
176			fork_inst: type) -> callable:
177			result = None
178			parents = get_parents(cls, classes_lizers)
179			if parents:
180			pname = get_class_name(parents[0], str.lower, fork_inst=fork_inst,
181			fully_qualified=True)
182			result = lizers[pname]
183			return result
184
185
186			def dumps(obj: object,
187			jdkwargs: Optional[Dict[str, object]] = None,
188			*args,
189			**kwargs) -> str:
190			"""
191			Extend ``json.dumps``, allowing any Python instance to be dumped to a
192			string. Any extra (keyword) arguments are passed on to ``json.dumps``.
193
194			:param obj: the object that is to be dumped to a string.
195			:param jdkwargs: extra keyword arguments for ``json.dumps`` (not
196			``jsons.dumps``!)
197			:param args: extra arguments for ``jsons.dumps``.
198			:param kwargs: Keyword arguments that are passed on through the
199			serialization process.
200			passed on to the serializer function.
201			:return: ``obj`` as a ``str``.
202			"""
203			jdkwargs = jdkwargs or {}
204			dumped = dump(obj, args, *kwargs)
205			return json.dumps(dumped, **jdkwargs)
206
207
208			def loads(str_: str,
209			cls: Optional[type] = None,
210			jdkwargs: Optional[Dict[str, object]] = None,
211			*args,
212			**kwargs) -> object:
213			"""
214			Extend ``json.loads``, allowing a string to be loaded into a dict or a
215			Python instance of type ``cls``. Any extra (keyword) arguments are passed
216			on to ``json.loads``.
217
218			:param str_: the string that is to be loaded.
219			:param cls: a matching class of which an instance should be returned.
220			:param jdkwargs: extra keyword arguments for ``json.loads`` (not
221			``jsons.loads``!)
222			:param args: extra arguments for ``jsons.loads``.
223			:param kwargs: extra keyword arguments for ``jsons.loads``.
224			:return: a JSON-type object (dict, str, list, etc.) or an instance of type
225			``cls`` if given.
226			"""
227			jdkwargs = jdkwargs or {}
228			try:
229			obj = json.loads(str_, **jdkwargs)
230			except JSONDecodeError as err:
231			raise DecodeError('Could not load a dict; the given string is not '
232			'valid JSON.', str_, cls, err)
233			else:
234			return load(obj, cls, args, *kwargs)
235
236
237			def dumpb(obj: object,
238			encoding: str = 'utf-8',
239			jdkwargs: Optional[Dict[str, object]] = None,
240			*args,
241			**kwargs) -> bytes:
242			"""
243			Extend ``json.dumps``, allowing any Python instance to be dumped to bytes.
244			Any extra (keyword) arguments are passed on to ``json.dumps``.
245
246			:param obj: the object that is to be dumped to bytes.
247			:param encoding: the encoding that is used to transform to bytes.
248			:param jdkwargs: extra keyword arguments for ``json.dumps`` (not
249			``jsons.dumps``!)
250			:param args: extra arguments for ``jsons.dumps``.
251			:param kwargs: Keyword arguments that are passed on through the
252			serialization process.
253			passed on to the serializer function.
254			:return: ``obj`` as ``bytes``.
255			"""
256			jdkwargs = jdkwargs or {}
257			dumped_dict = dump(obj, args, *kwargs)
258			dumped_str = json.dumps(dumped_dict, **jdkwargs)
259			return dumped_str.encode(encoding=encoding)
260
261
262			def loadb(bytes_: bytes,
263			cls: Optional[type] = None,
264			encoding: str = 'utf-8',
265			jdkwargs: Optional[Dict[str, object]] = None,
266			*args,
267			**kwargs) -> object:
268			"""
269			Extend ``json.loads``, allowing bytes to be loaded into a dict or a Python
270			instance of type ``cls``. Any extra (keyword) arguments are passed on to
271			``json.loads``.
272
273			:param bytes_: the bytes that are to be loaded.
274			:param cls: a matching class of which an instance should be returned.
275			:param encoding: the encoding that is used to transform from bytes.
276			:param jdkwargs: extra keyword arguments for ``json.loads`` (not
277			``jsons.loads``!)
278			:param args: extra arguments for ``jsons.loads``.
279			:param kwargs: extra keyword arguments for ``jsons.loads``.
280			:return: a JSON-type object (dict, str, list, etc.) or an instance of type
281			``cls`` if given.
282			"""
283			if not isinstance(bytes_, bytes):
284			raise DeserializationError('loadb accepts bytes only, "{}" was given'
285			.format(type(bytes_)), bytes_, cls)
286			jdkwargs = jdkwargs or {}
287			str_ = bytes_.decode(encoding=encoding)
288			return loads(str_, cls, jdkwargs=jdkwargs, args, *kwargs)
289
290
291			def set_serializer(func: callable,
292			cls: Union[type, Sequence[type]],
293			high_prio: bool = True,
294			fork_inst: type = StateHolder) -> None:
295			"""
296			Set a serializer function for the given type. You may override the default
297			behavior of ``jsons.load`` by setting a custom serializer.
298
299			The ``func`` argument must take one argument (i.e. the object that is to be
300			serialized) and also a ``kwargs`` parameter. For example:
301
302			>>> def func(obj, **kwargs):
303			... return dict()
304
305			You may ask additional arguments between ``cls`` and ``kwargs``.
306
307			:param func: the serializer function.
308			:param cls: the type or sequence of types this serializer can handle.
309			:param high_prio: determines the order in which is looked for the callable.
310			:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
311			:return: None.
312			"""
313			if isinstance(cls, Sequence):
314			for cls_ in cls:
315			set_serializer(func, cls_, high_prio, fork_inst)
316			elif cls:
317			index = 0 if high_prio else len(fork_inst._classes_serializers)
318			fork_inst._classes_serializers.insert(index, cls)
319			cls_name = get_class_name(cls, fork_inst=fork_inst,
320			fully_qualified=True)
321			fork_inst._serializers[cls_name.lower()] = func
322			else:
323			fork_inst._serializers['nonetype'] = func
324
325
326			def set_deserializer(func: callable,
327			cls: Union[type, Sequence[type]],
328			high_prio: bool = True,
329			fork_inst: type = StateHolder) -> None:
330			"""
331			Set a deserializer function for the given type. You may override the
332			default behavior of ``jsons.dump`` by setting a custom deserializer.
333
334			The ``func`` argument must take two arguments (i.e. the dict containing the
335			serialized values and the type that the values should be deserialized into)
336			and also a ``kwargs`` parameter. For example:
337
338			>>> def func(dict_, cls, **kwargs):
339			... return cls()
340
341			You may ask additional arguments between ``cls`` and ``kwargs``.
342
343			:param func: the deserializer function.
344			:param cls: the type or sequence of types this serializer can handle.
345			:param high_prio: determines the order in which is looked for the callable.
346			:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
347			:return: None.
348			"""
349			if isinstance(cls, Sequence):
350			for cls_ in cls:
351			set_deserializer(func, cls_, high_prio, fork_inst)
352			elif cls:
353			index = 0 if high_prio else len(fork_inst._classes_deserializers)
354			fork_inst._classes_deserializers.insert(index, cls)
355			cls_name = get_class_name(cls, fork_inst=fork_inst,
356			fully_qualified=True)
357			fork_inst._deserializers[cls_name.lower()] = func
358			else:
359			fork_inst._deserializers['nonetype'] = func
360
361
362			def suppress_warnings(
363			do_suppress: Optional[bool] = True,
364			fork_inst: Optional[type] = StateHolder):
365			"""
366			Suppress (or stop suppressing) warnings.
367			:param do_suppress: if ``True``, warnings will be suppressed from now on.
368			:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
369			:return: None.
370			"""
371			fork_inst._suppress_warnings = do_suppress
372
373
374			def announce_class(
375			cls: type,
376			cls_name: Optional[str] = None,
377			fork_inst: type = StateHolder):
378			"""
379			Announce the given cls to jsons to allow jsons to deserialize a verbose
380			dump into that class.
381			:param cls: the class that is to be announced.
382			:param cls_name: a custom name for that class.
383			:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
384			:return: None.
385			"""
386			cls_name = cls_name or get_class_name(cls, fully_qualified=True,
387			fork_inst=fork_inst)
388			fork_inst._announced_classes[cls] = cls_name
389			fork_inst._announced_classes[cls_name] = cls
390
391
392			def _check_and_get_cls_and_meta_hints(
393			json_obj: object,
394			cls: type,
395			fork_inst: type,
396			inferred_cls: bool) -> Tuple[type, Optional[dict]]:
397			# Check if json_obj is of a valid type and return the cls.
398			if type(json_obj) not in VALID_TYPES:
399			invalid_type = get_class_name(type(json_obj), fork_inst=fork_inst,
400			fully_qualified=True)
401			valid_types = [get_class_name(typ, fork_inst=fork_inst,
402			fully_qualified=True)
403			for typ in VALID_TYPES]
404			msg = ('Invalid type: "{}", only arguments of the following types are '
405			'allowed: {}'.format(invalid_type, ", ".join(valid_types)))
406			raise DeserializationError(msg, json_obj, cls)
407			if json_obj is None:
408			raise DeserializationError('Cannot load None with strict=True',
409			json_obj, cls)
410
411			cls_from_meta, meta = get_cls_and_meta(json_obj, fork_inst)
412			meta_hints = meta.get('classes', {}) if meta else {}
413			return determine_precedence(
414			cls, cls_from_meta, type(json_obj), inferred_cls), meta_hints
415

ramonhagenaars / jsons

Pull Request — master (#42)

jsons._main_impl._should_skip() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like