jsons._load_impl._do_load() - Code Metrics - Inspection of "Merge pull request #79 from haluzpav/better-type-e..." - ramonhagenaars/jsons - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 600917...5ea888 )

by Ramon

created 2019-12-21 21:44 UTC

jsons._load_impl._do_load() B

↳ Parent: jsons._load_impl

Complexity

Conditions

Size

Total Lines	22
Code Lines	20

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	20
dl	0
loc	22
rs	8.4666
c	0
b	0
f	0
cc	6
nop	5

"""
PRIVATE MODULE: do not import (from) it directly.

This module contains functionality for loading stuff from json.
"""
import json
from json import JSONDecodeError
from typing import Optional, Dict, Callable, Tuple, Any, Type
from jsons._cache import clear
from jsons._lizers_impl import get_deserializer
from jsons._validation import validate
from jsons.exceptions import DeserializationError, JsonsError, DecodeError
from jsons._common_impl import (
    StateHolder,
    get_cls_from_str,
    get_class_name,
    get_cls_and_meta,
    determine_precedence,
    VALID_TYPES,
    T,
    can_match_with_none
)


def load(
        json_obj: object,
        cls: Optional[Type[T]] = None,
        *,
        strict: bool = False,
        fork_inst: Optional[type] = StateHolder,
        attr_getters: Optional[Dict[str, Callable[[], object]]] = None,
        **kwargs) -> T:
    """
    Deserialize the given ``json_obj`` to an object of type ``cls``. If the
    contents of ``json_obj`` do not match the interface of ``cls``, a
    DeserializationError is raised.

    If ``json_obj`` contains a value that belongs to a custom class, there must
    be a type hint present for that value in ``cls`` to let this function know
    what type it should deserialize that value to.


    **Example**:

    >>> from typing import List
    >>> import jsons
    >>> class Person:
    ...     # No type hint required for name
    ...     def __init__(self, name):
    ...         self.name = name
    >>> class Family:
    ...     # Person is a custom class, use a type hint
    ...         def __init__(self, persons: List[Person]):
    ...             self.persons = persons
    >>> loaded = jsons.load({'persons': [{'name': 'John'}]}, Family)
    >>> loaded.persons[0].name
    'John'

    If no ``cls`` is given, a dict is simply returned, but contained values
    (e.g. serialized ``datetime`` values) are still deserialized.

    If `strict` mode is off and the type of `json_obj` exactly matches `cls`
    then `json_obj` is simply returned.

    :param json_obj: the dict that is to be deserialized.
    :param cls: a matching class of which an instance should be returned.
    :param strict: a bool to determine if the deserializer should be strict
    (i.e. fail on a partially deserialized `json_obj` or on `None`).
    :param fork_inst: if given, it uses this fork of ``JsonSerializable``.
    :param attr_getters: a ``dict`` that may hold callables that return values
    for certain attributes.
    :param kwargs: the keyword args are passed on to the deserializer function.
    :return: an instance of ``cls`` if given, a dict otherwise.
    """
    _check_for_none(json_obj, cls)
    if _should_skip(json_obj, cls, strict):
        validate(json_obj, cls, fork_inst)
        return json_obj
    if isinstance(cls, str):
        cls = get_cls_from_str(cls, json_obj, fork_inst)
    cls, meta_hints = _check_and_get_cls_and_meta_hints(
        json_obj, cls, fork_inst, kwargs.get('_inferred_cls', False))

    deserializer = get_deserializer(cls, fork_inst)

    # Is this the initial call or a nested?
    initial = kwargs.get('_initial', True)

    kwargs_ = {
        'strict': strict,
        'fork_inst': fork_inst,
        'attr_getters': attr_getters,
        'meta_hints': meta_hints,
        '_initial': False,
        **kwargs
    }

    return _do_load(json_obj, deserializer, cls, initial, **kwargs_)


def _do_load(json_obj: object,
             deserializer: callable,
             cls: type,
             initial: bool,
             **kwargs):
    cls_name = get_class_name(cls, fully_qualified=True)
    if deserializer is None:
        raise DeserializationError('No deserializer for type "{}"'.format(cls_name), json_obj, cls)
    try:
        result = deserializer(json_obj, cls, **kwargs)
        validate(result, cls, kwargs['fork_inst'])
    except Exception as err:
        clear()
        if isinstance(err, JsonsError):
            raise
        message = 'Could not deserialize value "{}" into "{}". {}'.format(json_obj, cls_name, err)
        raise DeserializationError(message, json_obj, cls)
    else:
        if initial:
            # Clear all lru caches right before returning the initial call.
            clear()
        return result


def loads(
        str_: str,
        cls: Optional[Type[T]] = None,
        jdkwargs: Optional[Dict[str, object]] = None,
        *args,
        **kwargs) -> T:
    """
    Extend ``json.loads``, allowing a string to be loaded into a dict or a
    Python instance of type ``cls``. Any extra (keyword) arguments are passed
    on to ``json.loads``.

    :param str_: the string that is to be loaded.
    :param cls: a matching class of which an instance should be returned.
    :param jdkwargs: extra keyword arguments for ``json.loads`` (not
    ``jsons.loads``!)
    :param args: extra arguments for ``jsons.loads``.
    :param kwargs: extra keyword arguments for ``jsons.loads``.
    :return: a JSON-type object (dict, str, list, etc.) or an instance of type
    ``cls`` if given.
    """
    jdkwargs = jdkwargs or {}
    try:
        obj = json.loads(str_, **jdkwargs)
    except JSONDecodeError as err:
        raise DecodeError('Could not load a dict; the given string is not '
                          'valid JSON.', str_, cls, err)
    else:
        return load(obj, cls, *args, **kwargs)


def loadb(
        bytes_: bytes,
        cls: Optional[Type[T]] = None,
        encoding: str = 'utf-8',
        jdkwargs: Optional[Dict[str, object]] = None,
        *args,
        **kwargs) -> T:
    """
    Extend ``json.loads``, allowing bytes to be loaded into a dict or a Python
    instance of type ``cls``. Any extra (keyword) arguments are passed on to
    ``json.loads``.

    :param bytes_: the bytes that are to be loaded.
    :param cls: a matching class of which an instance should be returned.
    :param encoding: the encoding that is used to transform from bytes.
    :param jdkwargs: extra keyword arguments for ``json.loads`` (not
    ``jsons.loads``!)
    :param args: extra arguments for ``jsons.loads``.
    :param kwargs: extra keyword arguments for ``jsons.loads``.
    :return: a JSON-type object (dict, str, list, etc.) or an instance of type
    ``cls`` if given.
    """
    if not isinstance(bytes_, bytes):
        raise DeserializationError('loadb accepts bytes only, "{}" was given'
                                   .format(type(bytes_)), bytes_, cls)
    jdkwargs = jdkwargs or {}
    str_ = bytes_.decode(encoding=encoding)
    return loads(str_, cls, jdkwargs=jdkwargs, *args, **kwargs)


def _check_and_get_cls_and_meta_hints(
        json_obj: object,
        cls: type,
        fork_inst: type,
        inferred_cls: bool) -> Tuple[type, Optional[dict]]:
    # Check if json_obj is of a valid type and return the cls.
    if type(json_obj) not in VALID_TYPES:
        invalid_type = get_class_name(type(json_obj), fork_inst=fork_inst,
                                      fully_qualified=True)
        valid_types = [get_class_name(typ, fork_inst=fork_inst,
                                      fully_qualified=True)
                       for typ in VALID_TYPES]
        msg = ('Invalid type: "{}", only arguments of the following types are '
               'allowed: {}'.format(invalid_type, ", ".join(valid_types)))
        raise DeserializationError(msg, json_obj, cls)

    cls_from_meta, meta = get_cls_and_meta(json_obj, fork_inst)
    meta_hints = meta.get('classes', {}) if meta else {}
    return determine_precedence(
        cls, cls_from_meta, type(json_obj), inferred_cls), meta_hints


def _should_skip(json_obj: object, cls: type, strict: bool):
    return (not strict and type(json_obj) == cls) or cls is Any


def _check_for_none(json_obj: object, cls: type):
    # Check if the json_obj is None and whether or not that is fine.
    if json_obj is None and not can_match_with_none(cls):
        cls_name = get_class_name(cls).lower()
        raise DeserializationError(
            message='NoneType cannot be deserialized into {}'.format(cls_name),
            source=json_obj,
            target=cls)


1			"""
2			PRIVATE MODULE: do not import (from) it directly.
3
4			This module contains functionality for loading stuff from json.
5			"""
6			import json
7			from json import JSONDecodeError
8			from typing import Optional, Dict, Callable, Tuple, Any, Type
9			from jsons._cache import clear
10			from jsons._lizers_impl import get_deserializer
11			from jsons._validation import validate
12			from jsons.exceptions import DeserializationError, JsonsError, DecodeError
13			from jsons._common_impl import (
14			StateHolder,
15			get_cls_from_str,
16			get_class_name,
17			get_cls_and_meta,
18			determine_precedence,
19			VALID_TYPES,
20			T,
21			can_match_with_none
22			)
23
24
25			def load(
26			json_obj: object,
27			cls: Optional[Type[T]] = None,
28			*,
29			strict: bool = False,
30			fork_inst: Optional[type] = StateHolder,
31			attr_getters: Optional[Dict[str, Callable[[], object]]] = None,
32			**kwargs) -> T:
33			"""
34			Deserialize the given ``json_obj`` to an object of type ``cls``. If the
35			contents of ``json_obj`` do not match the interface of ``cls``, a
36			DeserializationError is raised.
37
38			If ``json_obj`` contains a value that belongs to a custom class, there must
39			be a type hint present for that value in ``cls`` to let this function know
40			what type it should deserialize that value to.
41
42
43			Example:
44
45			>>> from typing import List
46			>>> import jsons
47			>>> class Person:
48			... # No type hint required for name
49			... def __init__(self, name):
50			... self.name = name
51			>>> class Family:
52			... # Person is a custom class, use a type hint
53			... def __init__(self, persons: List[Person]):
54			... self.persons = persons
55			>>> loaded = jsons.load({'persons': [{'name': 'John'}]}, Family)
56			>>> loaded.persons[0].name
57			'John'
58
59			If no ``cls`` is given, a dict is simply returned, but contained values
60			(e.g. serialized ``datetime`` values) are still deserialized.
61
62			If `strict` mode is off and the type of `json_obj` exactly matches `cls`
63			then `json_obj` is simply returned.
64
65			:param json_obj: the dict that is to be deserialized.
66			:param cls: a matching class of which an instance should be returned.
67			:param strict: a bool to determine if the deserializer should be strict
68			(i.e. fail on a partially deserialized `json_obj` or on `None`).
69			:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
70			:param attr_getters: a ``dict`` that may hold callables that return values
71			for certain attributes.
72			:param kwargs: the keyword args are passed on to the deserializer function.
73			:return: an instance of ``cls`` if given, a dict otherwise.
74			"""
75			_check_for_none(json_obj, cls)
76			if _should_skip(json_obj, cls, strict):
77			validate(json_obj, cls, fork_inst)
78			return json_obj
79			if isinstance(cls, str):
80			cls = get_cls_from_str(cls, json_obj, fork_inst)
81			cls, meta_hints = _check_and_get_cls_and_meta_hints(
82			json_obj, cls, fork_inst, kwargs.get('_inferred_cls', False))
83
84			deserializer = get_deserializer(cls, fork_inst)
85
86			# Is this the initial call or a nested?
87			initial = kwargs.get('_initial', True)
88
89			kwargs_ = {
90			'strict': strict,
91			'fork_inst': fork_inst,
92			'attr_getters': attr_getters,
93			'meta_hints': meta_hints,
94			'_initial': False,
95			**kwargs
96			}
97
98			return _do_load(json_obj, deserializer, cls, initial, **kwargs_)
99
100
101			def _do_load(json_obj: object,
102			deserializer: callable,
103			cls: type,
104			initial: bool,
105			**kwargs):
106			cls_name = get_class_name(cls, fully_qualified=True)
107			if deserializer is None:
108			raise DeserializationError('No deserializer for type "{}"'.format(cls_name), json_obj, cls)
109			try:
110			result = deserializer(json_obj, cls, **kwargs)
111			validate(result, cls, kwargs['fork_inst'])
112			except Exception as err:
113			clear()
114			if isinstance(err, JsonsError):
115			raise
116			message = 'Could not deserialize value "{}" into "{}". {}'.format(json_obj, cls_name, err)
117			raise DeserializationError(message, json_obj, cls)
118			else:
119			if initial:
120			# Clear all lru caches right before returning the initial call.
121			clear()
122			return result
123
124
125			def loads(
126			str_: str,
127			cls: Optional[Type[T]] = None,
128			jdkwargs: Optional[Dict[str, object]] = None,
129			*args,
130			**kwargs) -> T:
131			"""
132			Extend ``json.loads``, allowing a string to be loaded into a dict or a
133			Python instance of type ``cls``. Any extra (keyword) arguments are passed
134			on to ``json.loads``.
135
136			:param str_: the string that is to be loaded.
137			:param cls: a matching class of which an instance should be returned.
138			:param jdkwargs: extra keyword arguments for ``json.loads`` (not
139			``jsons.loads``!)
140			:param args: extra arguments for ``jsons.loads``.
141			:param kwargs: extra keyword arguments for ``jsons.loads``.
142			:return: a JSON-type object (dict, str, list, etc.) or an instance of type
143			``cls`` if given.
144			"""
145			jdkwargs = jdkwargs or {}
146			try:
147			obj = json.loads(str_, **jdkwargs)
148			except JSONDecodeError as err:
149			raise DecodeError('Could not load a dict; the given string is not '
150			'valid JSON.', str_, cls, err)
151			else:
152			return load(obj, cls, args, *kwargs)
153
154
155			def loadb(
156			bytes_: bytes,
157			cls: Optional[Type[T]] = None,
158			encoding: str = 'utf-8',
159			jdkwargs: Optional[Dict[str, object]] = None,
160			*args,
161			**kwargs) -> T:
162			"""
163			Extend ``json.loads``, allowing bytes to be loaded into a dict or a Python
164			instance of type ``cls``. Any extra (keyword) arguments are passed on to
165			``json.loads``.
166
167			:param bytes_: the bytes that are to be loaded.
168			:param cls: a matching class of which an instance should be returned.
169			:param encoding: the encoding that is used to transform from bytes.
170			:param jdkwargs: extra keyword arguments for ``json.loads`` (not
171			``jsons.loads``!)
172			:param args: extra arguments for ``jsons.loads``.
173			:param kwargs: extra keyword arguments for ``jsons.loads``.
174			:return: a JSON-type object (dict, str, list, etc.) or an instance of type
175			``cls`` if given.
176			"""
177			if not isinstance(bytes_, bytes):
178			raise DeserializationError('loadb accepts bytes only, "{}" was given'
179			.format(type(bytes_)), bytes_, cls)
180			jdkwargs = jdkwargs or {}
181			str_ = bytes_.decode(encoding=encoding)
182			return loads(str_, cls, jdkwargs=jdkwargs, args, *kwargs)
183
184
185			def _check_and_get_cls_and_meta_hints(
186			json_obj: object,
187			cls: type,
188			fork_inst: type,
189			inferred_cls: bool) -> Tuple[type, Optional[dict]]:
190			# Check if json_obj is of a valid type and return the cls.
191			if type(json_obj) not in VALID_TYPES:
192			invalid_type = get_class_name(type(json_obj), fork_inst=fork_inst,
193			fully_qualified=True)
194			valid_types = [get_class_name(typ, fork_inst=fork_inst,
195			fully_qualified=True)
196			for typ in VALID_TYPES]
197			msg = ('Invalid type: "{}", only arguments of the following types are '
198			'allowed: {}'.format(invalid_type, ", ".join(valid_types)))
199			raise DeserializationError(msg, json_obj, cls)
200
201			cls_from_meta, meta = get_cls_and_meta(json_obj, fork_inst)
202			meta_hints = meta.get('classes', {}) if meta else {}
203			return determine_precedence(
204			cls, cls_from_meta, type(json_obj), inferred_cls), meta_hints
205
206
207			def _should_skip(json_obj: object, cls: type, strict: bool):
208			return (not strict and type(json_obj) == cls) or cls is Any
209
210
211			def _check_for_none(json_obj: object, cls: type):
212			# Check if the json_obj is None and whether or not that is fine.
213			if json_obj is None and not can_match_with_none(cls):
214			cls_name = get_class_name(cls).lower()
215			raise DeserializationError(
216			message='NoneType cannot be deserialized into {}'.format(cls_name),
217			source=json_obj,
218			target=cls)
219

ramonhagenaars / jsons

Push — master ( 600917...5ea888 )

jsons._load_impl._do_load() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like