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
|
|
|
|