jsons._load_impl._do_load() - Code Metrics - Inspection of "Merge pull request #73 from ramonhagenaars/Release..." - ramonhagenaars/jsons - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( a0c5cf...fb555e )

by Ramon

created 2019-10-07 17:48 UTC

jsons._load_impl._do_load() A

↳ Parent: jsons._load_impl

Complexity

Conditions

Size

Total Lines	18
Code Lines	16

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

ramonhagenaars / jsons

Push — master ( a0c5cf...fb555e )

jsons._load_impl._do_load() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like