jsons.serializers.default_object._do_serialize() - Code Metrics - Inspection of "Merge pull request #117 from ramonhagenaars/Releas..." - ramonhagenaars/jsons - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 919d86...33db6d )

by Ramon

created 2020-10-09 20:07 UTC

jsons.serializers.default_object._do_serialize() B

↳ Parent: jsons.serializers.default_object

Complexity

Conditions

Size

Total Lines	44
Code Lines	40

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	40
dl	0
loc	44
rs	8.4533
c	0
b	0
f	0
cc	5
nop	12

How to fix Many Parameters

import inspect
from datetime import datetime, timezone
from inspect import isfunction
from typing import Optional, Callable, Union, MutableSequence, Tuple, Dict

from typish import get_type, get_mro

from jsons._cache import cached
from jsons._common_impl import get_class_name, META_ATTR, StateHolder
from jsons._compatibility_impl import get_type_hints
from jsons._datetime_impl import to_str
from jsons._dump_impl import dump
from jsons.classes import JsonSerializable
from jsons.classes.verbosity import Verbosity
from jsons.exceptions import SerializationError, RecursionDetectedError


def default_object_serializer(
        obj: object,
        cls: Optional[type] = None,
        *,
        key_transformer: Optional[Callable[[str], str]] = None,
        strip_nulls: bool = False,
        strip_privates: bool = False,
        strip_properties: bool = False,
        strip_class_variables: bool = False,
        strip_attr: Union[str, MutableSequence[str], Tuple[str]] = None,
        verbose: Union[Verbosity, bool] = False,
        strict: bool = False,
        fork_inst: Optional[type] = StateHolder,
        **kwargs) -> Optional[dict]:
    """
    Serialize the given ``obj`` to a dict. All values within ``obj`` are also
    serialized. If ``key_transformer`` is given, it will be used to transform
    the casing (e.g. snake_case) to a different format (e.g. camelCase).
    :param obj: the object that is to be serialized.
    :param cls: the type of the object that is to be dumped.
    :param key_transformer: a function that will be applied to all keys in the
    resulting dict.
    :param strip_nulls: if ``True`` the resulting dict will not contain null
    values.
    :param strip_privates: if ``True`` the resulting dict will not contain
    private attributes (i.e. attributes that start with an underscore).
    :param strip_properties: if ``True`` the resulting dict will not contain
    values from @properties.
    :param strip_class_variables: if ``True`` the resulting dict will not
    contain values from class variables.
    :param strip_attr: can be a name or a collection of names of attributes
    that are not to be included in the dump.
    :param verbose: if ``True`` the resulting dict will contain meta
    information (e.g. on how to deserialize).
    :param strict: a bool to determine if the serializer should be strict
    (i.e. only dumping stuff that is known to ``cls``).
    :param fork_inst: if given, it uses this fork of ``JsonSerializable``.
    :param kwargs: any keyword arguments that are to be passed to the
    serializer functions.
    :return: a Python dict holding the values
    of ``obj``.
    """
    if obj is None:
        return obj
    strip_attr = _normalize_strip_attr(strip_attr)
    if cls and strict:
        attributes = _get_attributes_from_class(
            cls, strip_privates, strip_properties, strip_class_variables,
            strip_attr, strict)
    else:
        attributes = _get_attributes_from_object(
            obj, strip_privates, strip_properties, strip_class_variables,
            strip_attr, strict)
        cls = obj.__class__
    verbose = Verbosity.from_value(verbose)
    kwargs_ = {
        **kwargs,
        'fork_inst': fork_inst,
        'verbose': verbose,
        'strict': strict,
        # Set a flag in kwargs to temporarily store -cls:
        '_store_cls': Verbosity.WITH_CLASS_INFO in verbose
    }

    result = _do_serialize(obj=obj,
                           cls=cls,
                           attributes=attributes,
                           kwargs=kwargs_,
                           key_transformer=key_transformer,
                           strip_nulls=strip_nulls,
                           strip_privates=strip_privates,
                           strip_properties=strip_properties,
                           strip_class_variables=strip_class_variables,
                           strip_attr=strip_attr,
                           strict=strict,
                           fork_inst=fork_inst)

    cls_name = get_class_name(cls, fully_qualified=True)
    if not kwargs.get('_store_cls'):
        result = _get_dict_with_meta(result, cls_name, verbose, fork_inst)
    return result


def _do_serialize(
        obj: object,
        cls: type,
        attributes: Dict[str, Optional[type]],
        kwargs: dict,
        key_transformer: Optional[Callable[[str], str]] = None,
        strip_nulls: bool = False,
        strip_privates: bool = False,
        strip_properties: bool = False,
        strip_class_variables: bool = False,
        strip_attr: Union[str, MutableSequence[str], Tuple[str]] = None,
        strict: bool = False,
        fork_inst: Optional[type] = StateHolder) -> Dict[str, object]:
    result = dict()
    for attr_name, cls_ in attributes.items():
        attr = obj.__getattribute__(attr_name)
        dumped_elem = None
        try:
            dumped_elem = dump(attr,
                               cls=cls_,
                               key_transformer=key_transformer,
                               strip_nulls=strip_nulls,
                               strip_privates=strip_privates,
                               strip_properties=strip_properties,
                               strip_class_variables=strip_class_variables,
                               strip_attr=strip_attr,
                               **kwargs)
            _store_cls_info(dumped_elem, attr, kwargs)
        except RecursionDetectedError:
            fork_inst._warn('Recursive structure detected in attribute "{}" '
                            'of object of type "{}", ignoring the attribute.'
                            .format(attr_name, get_class_name(cls)))
        except SerializationError as err:
            if strict:
                raise
            else:
                fork_inst._warn('Failed to dump attribute "{}" of object of '
                                'type "{}". Reason: {}. Ignoring the '
                                'attribute.'
                                .format(attr, get_class_name(cls), err.message))
                break
        _add_dumped_elem(result, attr_name, dumped_elem,
                         strip_nulls, key_transformer)
    return result


def _add_dumped_elem(
        result: dict,
        attr_name: str,
        dumped_elem: object,
        strip_nulls: bool,
        key_transformer: Optional[Callable[[str], str]]):
    if not (strip_nulls and dumped_elem is None):
        if key_transformer:
            attr_name = key_transformer(attr_name)
        result[attr_name] = dumped_elem


def _normalize_strip_attr(strip_attr) -> tuple:
    # Make sure that strip_attr is always a tuple.
    strip_attr = strip_attr or tuple()
    if (not isinstance(strip_attr, MutableSequence)
            and not isinstance(strip_attr, tuple)):
        strip_attr = (strip_attr,)
    return strip_attr


@cached
def _get_attributes_from_class(
        cls: type,
        strip_privates: bool,
        strip_properties: bool,
        strip_class_variables: bool,
        strip_attr: tuple,
        strict: bool) -> Dict[str, Optional[type]]:
    # Get the attributes that are known in the class.
    attributes_and_types = _get_attributes_and_types(cls, strict)
    return _filter_attributes(cls, attributes_and_types, strip_privates,
                              strip_properties, strip_class_variables,
                              strip_attr)


def _get_attributes_from_object(
        obj: object,
        strip_privates: bool,
        strip_properties: bool,
        strip_class_variables: bool,
        strip_attr: tuple,
        strict: bool) -> Dict[str, Optional[type]]:
    # Get the attributes that are known in the object.
    cls = obj.__class__
    attributes_and_types = _get_attributes_and_types(cls, strict)
    attributes = {attr: attributes_and_types.get(attr, None)
                  for attr in dir(obj)}
    return _filter_attributes(cls, attributes, strip_privates,
                              strip_properties, strip_class_variables,
                              strip_attr)


@cached
def _get_attributes_and_types(cls: type,
                              strict: bool) -> Dict[str, Optional[type]]:
    if '__slots__' in cls.__dict__:
        attributes = {attr: None for attr in cls.__slots__}
    elif hasattr(cls, '__annotations__'):
        attributes = get_type_hints(cls)
    elif strict:
        hints = get_type_hints(cls.__init__)
        attributes = {k: hints[k] for k in hints if k != 'self'}
    else:
        attributes = {}

    # Add properties and class variables.
    props, class_vars = _get_class_props(cls)
    for elem in props + class_vars:
        attributes[elem] = None

    return attributes


def _filter_attributes(
        cls: type,
        attributes: Dict[str, Optional[type]],
        strip_privates: bool,
        strip_properties: bool,
        strip_class_variables: bool,
        strip_attr: tuple) -> Dict[str, Optional[type]]:
    # Filter the given attributes with the given preferences.
    strip_attr = strip_attr + _ABC_ATTRS
    excluded_elems = dir(JsonSerializable)
    props, other_cls_vars = _get_class_props(cls)

    return {attr: type_ for attr, type_ in attributes.items()
            if not attr.startswith('__')
            and not (strip_privates and attr.startswith('_'))
            and not (strip_properties and attr in props)
            and not (strip_class_variables and attr in other_cls_vars)
            and attr not in strip_attr
            and attr != 'json'
            and not isfunction(getattr(cls, attr, None))
            and attr not in excluded_elems
            and not _is_innerclass(attr, cls)}


def _get_class_props(cls: type) -> Tuple[list, list]:
    props = []
    other_cls_vars = []
    for n, v in _get_complete_class_dict(cls).items():
        list_to_append = props if isinstance(v, property) else other_cls_vars
        list_to_append.append(n)
    return props, other_cls_vars


def _get_complete_class_dict(cls: type) -> dict:
    cls_dict = {}
    # Loop reversed so values of sub-classes override those of super-classes.
    for cls_or_elder in reversed(get_mro(cls)):
        cls_dict.update(cls_or_elder.__dict__)
    return cls_dict


def _get_dict_with_meta(
        obj: dict,
        cls_name: str,
        verbose: Verbosity,
        fork_inst: type) -> dict:
    # This function will add a -meta section to the given obj (provided that
    # the given obj has -cls attributes for all children).
    if verbose is Verbosity.WITH_NOTHING:
        return obj

    obj[META_ATTR] = {}
    if Verbosity.WITH_CLASS_INFO in verbose:
        collection_of_types = {}
        _fill_collection_of_types(obj, cls_name, '/', collection_of_types)
        collection_of_types['/'] = cls_name
        obj[META_ATTR]['classes'] = collection_of_types
    if Verbosity.WITH_DUMP_TIME in verbose:
        dump_time = to_str(datetime.now(tz=timezone.utc), True, fork_inst)
        obj[META_ATTR]['dump_time'] = dump_time
    return obj


def _fill_collection_of_types(
        obj_: dict,
        cls_name_: Optional[str],
        prefix: str,
        collection_of_types_: dict) -> str:
    # This function loops through obj_ to fill collection_of_types_ with the
    # class names. All of the -cls attributes are removed in the process.
    cls_name_ = _get_class_name_and_strip_cls(cls_name_, obj_)
    for attr in obj_:
        if attr != META_ATTR and isinstance(obj_[attr], dict):
            attr_class = _fill_collection_of_types(obj_[attr],
                                                   None,
                                                   prefix + attr + '/',
                                                   collection_of_types_)
            collection_of_types_[prefix + attr] = attr_class
    return cls_name_


def _get_class_name_and_strip_cls(cls_name: Optional[str], obj: dict) -> str:
    result = cls_name
    if not cls_name and '-cls' in obj:
        result = obj['-cls']
    if '-cls' in obj:
        del obj['-cls']
    return result


def _store_cls_info(result: object, original_obj: dict, kwargs):
    if kwargs.get('_store_cls', None) and isinstance(result, dict):
        cls = get_type(original_obj)
        if cls.__module__ == 'typing':
            cls_name = repr(cls)
        else:
            cls_name = get_class_name(cls, fully_qualified=True)
        result['-cls'] = cls_name


def _is_innerclass(attr: str, cls: type) -> bool:
    attr_obj = getattr(cls, attr, None)
    return (isinstance(attr_obj, type)
            and inspect.getsource(attr_obj) in inspect.getsource(cls))


_ABC_ATTRS = ('_abc_registry', '_abc_cache', '_abc_negative_cache',
              '_abc_negative_cache_version', '_abc_impl')


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._cache import cached
9			from jsons._common_impl import get_class_name, META_ATTR, StateHolder
10			from jsons._compatibility_impl import get_type_hints
11			from jsons._datetime_impl import to_str
12			from jsons._dump_impl import dump
13			from jsons.classes import JsonSerializable
14			from jsons.classes.verbosity import Verbosity
15			from jsons.exceptions import SerializationError, RecursionDetectedError
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			if obj is None:
61			return obj
62			strip_attr = _normalize_strip_attr(strip_attr)
63			if cls and strict:
64			attributes = _get_attributes_from_class(
65			cls, strip_privates, strip_properties, strip_class_variables,
66			strip_attr, strict)
67			else:
68			attributes = _get_attributes_from_object(
69			obj, strip_privates, strip_properties, strip_class_variables,
70			strip_attr, strict)
71			cls = obj.__class__
72			verbose = Verbosity.from_value(verbose)
73			kwargs_ = {
74			**kwargs,
75			'fork_inst': fork_inst,
76			'verbose': verbose,
77			'strict': strict,
78			# Set a flag in kwargs to temporarily store -cls:
79			'_store_cls': Verbosity.WITH_CLASS_INFO in verbose
80			}
81
82			result = _do_serialize(obj=obj,
83			cls=cls,
84			attributes=attributes,
85			kwargs=kwargs_,
86			key_transformer=key_transformer,
87			strip_nulls=strip_nulls,
88			strip_privates=strip_privates,
89			strip_properties=strip_properties,
90			strip_class_variables=strip_class_variables,
91			strip_attr=strip_attr,
92			strict=strict,
93			fork_inst=fork_inst)
94
95			cls_name = get_class_name(cls, fully_qualified=True)
96			if not kwargs.get('_store_cls'):
97			result = _get_dict_with_meta(result, cls_name, verbose, fork_inst)
98			return result
99
100
101			def _do_serialize(
102			obj: object,
103			cls: type,
104			attributes: Dict[str, Optional[type]],
105			kwargs: dict,
106			key_transformer: Optional[Callable[[str], str]] = None,
107			strip_nulls: bool = False,
108			strip_privates: bool = False,
109			strip_properties: bool = False,
110			strip_class_variables: bool = False,
111			strip_attr: Union[str, MutableSequence[str], Tuple[str]] = None,
112			strict: bool = False,
113			fork_inst: Optional[type] = StateHolder) -> Dict[str, object]:
114			result = dict()
115			for attr_name, cls_ in attributes.items():
116			attr = obj.__getattribute__(attr_name)
117			dumped_elem = None
118			try:
119			dumped_elem = dump(attr,
120			cls=cls_,
121			key_transformer=key_transformer,
122			strip_nulls=strip_nulls,
123			strip_privates=strip_privates,
124			strip_properties=strip_properties,
125			strip_class_variables=strip_class_variables,
126			strip_attr=strip_attr,
127			**kwargs)
128			_store_cls_info(dumped_elem, attr, kwargs)
129			except RecursionDetectedError:
130			fork_inst._warn('Recursive structure detected in attribute "{}" '
131			'of object of type "{}", ignoring the attribute.'
132			.format(attr_name, get_class_name(cls)))
133			except SerializationError as err:
134			if strict:
135			raise
136			else:
137			fork_inst._warn('Failed to dump attribute "{}" of object of '
138			'type "{}". Reason: {}. Ignoring the '
139			'attribute.'
140			.format(attr, get_class_name(cls), err.message))
141			break
142			_add_dumped_elem(result, attr_name, dumped_elem,
143			strip_nulls, key_transformer)
144			return result
145
146
147			def _add_dumped_elem(
148			result: dict,
149			attr_name: str,
150			dumped_elem: object,
151			strip_nulls: bool,
152			key_transformer: Optional[Callable[[str], str]]):
153			if not (strip_nulls and dumped_elem is None):
154			if key_transformer:
155			attr_name = key_transformer(attr_name)
156			result[attr_name] = dumped_elem
157
158
159			def _normalize_strip_attr(strip_attr) -> tuple:
160			# Make sure that strip_attr is always a tuple.
161			strip_attr = strip_attr or tuple()
162			if (not isinstance(strip_attr, MutableSequence)
163			and not isinstance(strip_attr, tuple)):
164			strip_attr = (strip_attr,)
165			return strip_attr
166
167
168			@cached
169			def _get_attributes_from_class(
170			cls: type,
171			strip_privates: bool,
172			strip_properties: bool,
173			strip_class_variables: bool,
174			strip_attr: tuple,
175			strict: bool) -> Dict[str, Optional[type]]:
176			# Get the attributes that are known in the class.
177			attributes_and_types = _get_attributes_and_types(cls, strict)
178			return _filter_attributes(cls, attributes_and_types, strip_privates,
179			strip_properties, strip_class_variables,
180			strip_attr)
181
182
183			def _get_attributes_from_object(
184			obj: object,
185			strip_privates: bool,
186			strip_properties: bool,
187			strip_class_variables: bool,
188			strip_attr: tuple,
189			strict: bool) -> Dict[str, Optional[type]]:
190			# Get the attributes that are known in the object.
191			cls = obj.__class__
192			attributes_and_types = _get_attributes_and_types(cls, strict)
193			attributes = {attr: attributes_and_types.get(attr, None)
194			for attr in dir(obj)}
195			return _filter_attributes(cls, attributes, strip_privates,
196			strip_properties, strip_class_variables,
197			strip_attr)
198
199
200			@cached
201			def _get_attributes_and_types(cls: type,
202			strict: bool) -> Dict[str, Optional[type]]:
203			if '__slots__' in cls.__dict__:
204			attributes = {attr: None for attr in cls.__slots__}
205			elif hasattr(cls, '__annotations__'):
206			attributes = get_type_hints(cls)
207			elif strict:
208			hints = get_type_hints(cls.__init__)
209			attributes = {k: hints[k] for k in hints if k != 'self'}
210			else:
211			attributes = {}
212
213			# Add properties and class variables.
214			props, class_vars = _get_class_props(cls)
215			for elem in props + class_vars:
216			attributes[elem] = None
217
218			return attributes
219
220
221			def _filter_attributes(
222			cls: type,
223			attributes: Dict[str, Optional[type]],
224			strip_privates: bool,
225			strip_properties: bool,
226			strip_class_variables: bool,
227			strip_attr: tuple) -> Dict[str, Optional[type]]:
228			# Filter the given attributes with the given preferences.
229			strip_attr = strip_attr + _ABC_ATTRS
230			excluded_elems = dir(JsonSerializable)
231			props, other_cls_vars = _get_class_props(cls)
232
233			return {attr: type_ for attr, type_ in attributes.items()
234			if not attr.startswith('__')
235			and not (strip_privates and attr.startswith('_'))
236			and not (strip_properties and attr in props)
237			and not (strip_class_variables and attr in other_cls_vars)
238			and attr not in strip_attr
239			and attr != 'json'
240			and not isfunction(getattr(cls, attr, None))
241			and attr not in excluded_elems
242			and not _is_innerclass(attr, cls)}
243
244
245			def _get_class_props(cls: type) -> Tuple[list, list]:
246			props = []
247			other_cls_vars = []
248			for n, v in _get_complete_class_dict(cls).items():
249			list_to_append = props if isinstance(v, property) else other_cls_vars
250			list_to_append.append(n)
251			return props, other_cls_vars
252
253
254			def _get_complete_class_dict(cls: type) -> dict:
255			cls_dict = {}
256			# Loop reversed so values of sub-classes override those of super-classes.
257			for cls_or_elder in reversed(get_mro(cls)):
258			cls_dict.update(cls_or_elder.__dict__)
259			return cls_dict
260
261
262			def _get_dict_with_meta(
263			obj: dict,
264			cls_name: str,
265			verbose: Verbosity,
266			fork_inst: type) -> dict:
267			# This function will add a -meta section to the given obj (provided that
268			# the given obj has -cls attributes for all children).
269			if verbose is Verbosity.WITH_NOTHING:
270			return obj
271
272			obj[META_ATTR] = {}
273			if Verbosity.WITH_CLASS_INFO in verbose:
274			collection_of_types = {}
275			_fill_collection_of_types(obj, cls_name, '/', collection_of_types)
276			collection_of_types['/'] = cls_name
277			obj[META_ATTR]['classes'] = collection_of_types
278			if Verbosity.WITH_DUMP_TIME in verbose:
279			dump_time = to_str(datetime.now(tz=timezone.utc), True, fork_inst)
280			obj[META_ATTR]['dump_time'] = dump_time
281			return obj
282
283
284			def _fill_collection_of_types(
285			obj_: dict,
286			cls_name_: Optional[str],
287			prefix: str,
288			collection_of_types_: dict) -> str:
289			# This function loops through obj_ to fill collection_of_types_ with the
290			# class names. All of the -cls attributes are removed in the process.
291			cls_name_ = _get_class_name_and_strip_cls(cls_name_, obj_)
292			for attr in obj_:
293			if attr != META_ATTR and isinstance(obj_[attr], dict):
294			attr_class = _fill_collection_of_types(obj_[attr],
295			None,
296			prefix + attr + '/',
297			collection_of_types_)
298			collection_of_types_[prefix + attr] = attr_class
299			return cls_name_
300
301
302			def _get_class_name_and_strip_cls(cls_name: Optional[str], obj: dict) -> str:
303			result = cls_name
304			if not cls_name and '-cls' in obj:
305			result = obj['-cls']
306			if '-cls' in obj:
307			del obj['-cls']
308			return result
309
310
311			def _store_cls_info(result: object, original_obj: dict, kwargs):
312			if kwargs.get('_store_cls', None) and isinstance(result, dict):
313			cls = get_type(original_obj)
314			if cls.__module__ == 'typing':
315			cls_name = repr(cls)
316			else:
317			cls_name = get_class_name(cls, fully_qualified=True)
318			result['-cls'] = cls_name
319
320
321			def _is_innerclass(attr: str, cls: type) -> bool:
322			attr_obj = getattr(cls, attr, None)
323			return (isinstance(attr_obj, type)
324			and inspect.getsource(attr_obj) in inspect.getsource(cls))
325
326
327			_ABC_ATTRS = ('_abc_registry', '_abc_cache', '_abc_negative_cache',
328			'_abc_negative_cache_version', '_abc_impl')
329

ramonhagenaars / jsons

Push — master ( 919d86...33db6d )

jsons.serializers.default_object._do_serialize() B

Complexity

Size

Duplication

Importance

How to fix Many Parameters

Many Parameters

Duplication Side-by-Side

Filter issues like