Passed
Pull Request — master (#38)
by Ramon
01:06
created

jsons._main_impl._get_cls_and_meta()   A

Complexity

Conditions 3

Size

Total Lines 8
Code Lines 8

Duplication

Lines 0
Ratio 0 %

Importance

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