jsons.serializers.default_object._do_serialize()   B
last analyzed

Complexity

Conditions 5

Size

Total Lines 48
Code Lines 44

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
eloc 44
dl 0
loc 48
rs 8.3573
c 0
b 0
f 0
cc 5
nop 12

How to fix   Many Parameters   

Many Parameters

Methods with many parameters are not only hard to understand, but their parameters also often become inconsistent when you need more, or different data.

There are several approaches to avoid long parameter lists:

1
import inspect
2
from datetime import datetime, timezone
3
from inspect import isfunction
4
from typing import Optional, Callable, Union, MutableSequence, Tuple, Dict
5
6
from typish import get_type, get_mro
7
8
from jsons import get_serializer, announce_class
9
from jsons._cache import cached
10
from jsons._common_impl import get_class_name, META_ATTR, StateHolder
11
from jsons._compatibility_impl import get_type_hints
12
from jsons._datetime_impl import to_str
13
from jsons.classes import JsonSerializable
14
from jsons.classes.verbosity import Verbosity
15
from jsons.exceptions import SerializationError
16
17
18
def default_object_serializer(
19
        obj: object,
20
        cls: Optional[type] = None,
21
        *,
22
        key_transformer: Optional[Callable[[str], str]] = None,
23
        strip_nulls: bool = False,
24
        strip_privates: bool = False,
25
        strip_properties: bool = False,
26
        strip_class_variables: bool = False,
27
        strip_attr: Union[str, MutableSequence[str], Tuple[str]] = None,
28
        verbose: Union[Verbosity, bool] = False,
29
        strict: bool = False,
30
        fork_inst: Optional[type] = StateHolder,
31
        **kwargs) -> Optional[dict]:
32
    """
33
    Serialize the given ``obj`` to a dict. All values within ``obj`` are also
34
    serialized. If ``key_transformer`` is given, it will be used to transform
35
    the casing (e.g. snake_case) to a different format (e.g. camelCase).
36
    :param obj: the object that is to be serialized.
37
    :param cls: the type of the object that is to be dumped.
38
    :param key_transformer: a function that will be applied to all keys in the
39
    resulting dict.
40
    :param strip_nulls: if ``True`` the resulting dict will not contain null
41
    values.
42
    :param strip_privates: if ``True`` the resulting dict will not contain
43
    private attributes (i.e. attributes that start with an underscore).
44
    :param strip_properties: if ``True`` the resulting dict will not contain
45
    values from @properties.
46
    :param strip_class_variables: if ``True`` the resulting dict will not
47
    contain values from class variables.
48
    :param strip_attr: can be a name or a collection of names of attributes
49
    that are not to be included in the dump.
50
    :param verbose: if ``True`` the resulting dict will contain meta
51
    information (e.g. on how to deserialize).
52
    :param strict: a bool to determine if the serializer should be strict
53
    (i.e. only dumping stuff that is known to ``cls``).
54
    :param fork_inst: if given, it uses this fork of ``JsonSerializable``.
55
    :param kwargs: any keyword arguments that are to be passed to the
56
    serializer functions.
57
    :return: a Python dict holding the values
58
    of ``obj``.
59
    """
60
    strip_attr = _normalize_strip_attr(strip_attr)
61
    if cls and strict:
62
        attributes = _get_attributes_from_class(
63
            cls, strip_privates, strip_properties, strip_class_variables,
64
            strip_attr, strict)
65
    else:
66
        attributes = _get_attributes_from_object(
67
            obj, strip_privates, strip_properties, strip_class_variables,
68
            strip_attr, strict)
69
        cls = obj.__class__
70
71
    verbose = Verbosity.from_value(verbose)
72
    kwargs_ = {
73
        **kwargs,
74
        'fork_inst': fork_inst,
75
        'verbose': verbose,
76
        'strict': strict,
77
        # Set a flag in kwargs to temporarily store -cls:
78
        '_store_cls': Verbosity.WITH_CLASS_INFO in verbose
79
    }
80
81
    result = _do_serialize(obj=obj,
82
                           cls=cls,
83
                           attributes=attributes,
84
                           kwargs=kwargs_,
85
                           key_transformer=key_transformer,
86
                           strip_nulls=strip_nulls,
87
                           strip_privates=strip_privates,
88
                           strip_properties=strip_properties,
89
                           strip_class_variables=strip_class_variables,
90
                           strip_attr=strip_attr,
91
                           strict=strict,
92
                           fork_inst=fork_inst)
93
94
    cls_name = get_class_name(cls, fully_qualified=True)
95
    if not kwargs.get('_store_cls'):
96
        result = _get_dict_with_meta(result, cls_name, verbose, fork_inst)
97
    return result
98
99
100
def _do_serialize(
101
        obj: object,
102
        cls: type,
103
        attributes: Dict[str, Optional[type]],
104
        kwargs: dict,
105
        key_transformer: Optional[Callable[[str], str]] = None,
106
        strip_nulls: bool = False,
107
        strip_privates: bool = False,
108
        strip_properties: bool = False,
109
        strip_class_variables: bool = False,
110
        strip_attr: Union[str, MutableSequence[str], Tuple[str]] = None,
111
        strict: bool = False,
112
        fork_inst: Optional[type] = StateHolder) -> Dict[str, object]:
113
    result = dict()
114
    is_attrs_cls = getattr(cls, '__attrs_attrs__', None) is not None
115
    make_attributes_public = is_attrs_cls and not strip_privates
116
    for attr_name, cls_ in attributes.items():
117
        attr = getattr(obj, attr_name)
118
        attr_type = cls_ or type(attr)
119
        announce_class(attr_type, fork_inst=fork_inst)
120
        serializer = get_serializer(attr_type, fork_inst)
121
        try:
122
            dumped_elem = serializer(attr,
123
                                     cls=cls_,
124
                                     key_transformer=key_transformer,
125
                                     strip_nulls=strip_nulls,
126
                                     strip_privates=strip_privates,
127
                                     strip_properties=strip_properties,
128
                                     strip_class_variables=strip_class_variables,
129
                                     strip_attr=strip_attr,
130
                                     **kwargs)
131
            _store_cls_info(dumped_elem, attr, kwargs)
132
        except Exception as err:
133
            if strict:
134
                raise SerializationError(message=err.args[0]) from err
135
            else:
136
                fork_inst._warn('Failed to dump attribute "{}" of object of '
137
                                'type "{}". Reason: {}. Ignoring the '
138
                                'attribute.'
139
                                .format(attr, get_class_name(cls), err.args[0]),
140
                                'attribute-not-serialized')
141
                break
142
143
        if make_attributes_public:
144
            attr_name = attr_name.lstrip('_')
145
        _add_dumped_elem(result, attr_name, dumped_elem,
146
                         strip_nulls, key_transformer)
147
    return result
148
149
150
def _add_dumped_elem(
151
        result: dict,
152
        attr_name: str,
153
        dumped_elem: object,
154
        strip_nulls: bool,
155
        key_transformer: Optional[Callable[[str], str]]):
156
    if not (strip_nulls and dumped_elem is None):
157
        if key_transformer:
158
            attr_name = key_transformer(attr_name)
159
        result[attr_name] = dumped_elem
160
161
162
def _normalize_strip_attr(strip_attr) -> tuple:
163
    # Make sure that strip_attr is always a tuple.
164
    strip_attr = strip_attr or tuple()
165
    if (not isinstance(strip_attr, MutableSequence)
166
            and not isinstance(strip_attr, tuple)):
167
        strip_attr = (strip_attr,)
168
    return strip_attr
169
170
171
@cached
172
def _get_attributes_from_class(
173
        cls: type,
174
        strip_privates: bool,
175
        strip_properties: bool,
176
        strip_class_variables: bool,
177
        strip_attr: tuple,
178
        strict: bool) -> Dict[str, Optional[type]]:
179
    # Get the attributes that are known in the class.
180
    attributes_and_types = _get_attributes_and_types(cls, strict)
181
    return _filter_attributes(cls, attributes_and_types, strip_privates,
182
                              strip_properties, strip_class_variables,
183
                              strip_attr)
184
185
186
def _get_attributes_from_object(
187
        obj: object,
188
        strip_privates: bool,
189
        strip_properties: bool,
190
        strip_class_variables: bool,
191
        strip_attr: tuple,
192
        strict: bool) -> Dict[str, Optional[type]]:
193
    # Get the attributes that are known in the object.
194
    cls = obj.__class__
195
    attributes_and_types = _get_attributes_and_types(cls, strict)
196
    attributes = {attr: attributes_and_types.get(attr, None)
197
                  for attr in dir(obj)}
198
    return _filter_attributes(cls, attributes, strip_privates,
199
                              strip_properties, strip_class_variables,
200
                              strip_attr)
201
202
203
@cached
204
def _get_attributes_and_types(cls: type,
205
                              strict: bool) -> Dict[str, Optional[type]]:
206
    if '__slots__' in cls.__dict__:
207
        attributes = {attr: None for attr in cls.__slots__}
208
    elif hasattr(cls, '__annotations__'):
209
        attributes = get_type_hints(cls)
210
    elif strict:
211
        hints = get_type_hints(cls.__init__)
212
        attributes = {k: hints[k] for k in hints if k != 'self'}
213
    else:
214
        attributes = {}
215
216
    # Add properties and class variables.
217
    props, class_vars = _get_class_props(cls)
218
    for elem in props + class_vars:
219
        attributes[elem] = None
220
221
    return attributes
222
223
224
def _filter_attributes(
225
        cls: type,
226
        attributes: Dict[str, Optional[type]],
227
        strip_privates: bool,
228
        strip_properties: bool,
229
        strip_class_variables: bool,
230
        strip_attr: tuple) -> Dict[str, Optional[type]]:
231
    # Filter the given attributes with the given preferences.
232
    strip_attr = strip_attr + _ABC_ATTRS
233
    excluded_elems = dir(JsonSerializable)
234
    props, other_cls_vars = _get_class_props(cls)
235
236
    return {attr: type_ for attr, type_ in attributes.items()
237
            if not attr.startswith('__')
238
            and not (strip_privates and attr.startswith('_'))
239
            and not (strip_properties and attr in props)
240
            and not (strip_class_variables and attr in other_cls_vars)
241
            and attr not in strip_attr
242
            and attr != 'json'
243
            and not inspect.ismethod(getattr(cls, attr, None))
244
            and not isfunction(getattr(cls, attr, None))
245
            and attr not in excluded_elems
246
            and not _is_innerclass(attr, cls)}
247
248
249
@cached
250
def _get_class_props(cls: type) -> Tuple[list, list]:
251
    props = []
252
    other_cls_vars = []
253
    for n, v in _get_complete_class_dict(cls).items():
254
        list_to_append = props if isinstance(v, property) else other_cls_vars
255
        list_to_append.append(n)
256
    return props, other_cls_vars
257
258
259
def _get_complete_class_dict(cls: type) -> dict:
260
    cls_dict = {}
261
    # Loop reversed so values of sub-classes override those of super-classes.
262
    for cls_or_elder in reversed(get_mro(cls)):
263
        cls_dict.update(cls_or_elder.__dict__)
264
    return cls_dict
265
266
267
def _get_dict_with_meta(
268
        obj: dict,
269
        cls_name: str,
270
        verbose: Verbosity,
271
        fork_inst: type) -> dict:
272
    # This function will add a -meta section to the given obj (provided that
273
    # the given obj has -cls attributes for all children).
274
    if verbose is Verbosity.WITH_NOTHING:
275
        return obj
276
277
    obj[META_ATTR] = {}
278
    if Verbosity.WITH_CLASS_INFO in verbose:
279
        collection_of_types = {}
280
        _fill_collection_of_types(obj, cls_name, '/', collection_of_types)
281
        collection_of_types['/'] = cls_name
282
        obj[META_ATTR]['classes'] = collection_of_types
283
    if Verbosity.WITH_DUMP_TIME in verbose:
284
        dump_time = to_str(datetime.now(tz=timezone.utc), True, fork_inst)
285
        obj[META_ATTR]['dump_time'] = dump_time
286
    return obj
287
288
289
def _fill_collection_of_types(
290
        obj_: dict,
291
        cls_name_: Optional[str],
292
        prefix: str,
293
        collection_of_types_: dict) -> str:
294
    # This function loops through obj_ to fill collection_of_types_ with the
295
    # class names. All of the -cls attributes are removed in the process.
296
    cls_name_ = _get_class_name_and_strip_cls(cls_name_, obj_)
297
    for attr in obj_:
298
        if attr != META_ATTR and isinstance(obj_[attr], dict):
299
            attr_class = _fill_collection_of_types(obj_[attr],
300
                                                   None,
301
                                                   prefix + attr + '/',
302
                                                   collection_of_types_)
303
            collection_of_types_[prefix + attr] = attr_class
304
    return cls_name_
305
306
307
def _get_class_name_and_strip_cls(cls_name: Optional[str], obj: dict) -> str:
308
    result = cls_name
309
    if not cls_name and '-cls' in obj:
310
        result = obj['-cls']
311
    if '-cls' in obj:
312
        del obj['-cls']
313
    return result
314
315
316
def _store_cls_info(result: object, original_obj: dict, kwargs):
317
    if kwargs.get('_store_cls', None) and isinstance(result, dict):
318
        cls = get_type(original_obj)
319
        if cls.__module__ == 'typing':
320
            cls_name = repr(cls)
321
        else:
322
            cls_name = get_class_name(cls, fully_qualified=True)
323
        result['-cls'] = cls_name
324
325
326
@cached
327
def _is_innerclass(attr: str, cls: type) -> bool:
328
    attr_obj = getattr(cls, attr, None)
329
    return (isinstance(attr_obj, type)
330
            and inspect.getsource(attr_obj) in inspect.getsource(cls))
331
332
333
_ABC_ATTRS = ('_abc_registry', '_abc_cache', '_abc_negative_cache',
334
              '_abc_negative_cache_version', '_abc_impl')
335