jsons.serializers.default_object._is_innerclass() - 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._is_innerclass() A

↳ Parent: jsons.serializers.default_object

Complexity

Conditions

Size

Total Lines	4
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	4
dl	0
loc	4
rs	10
c	0
b	0
f	0
cc	1
nop	2

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._is_innerclass() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like