Passed
Pull Request — master (#17)
by Ramon
50s
created

jsons._main_impl.JsonSerializable.fork()   A

Complexity

Conditions 1

Size

Total Lines 18
Code Lines 12

Duplication

Lines 0
Ratio 0 %

Importance

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