Completed
Push — master ( 008c50...892606 )
by Ramon
11s
created

jsons._main_impl._should_skip()   A

Complexity

Conditions 4

Size

Total Lines 3
Code Lines 3

Duplication

Lines 0
Ratio 0 %

Importance

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