Completed
Push — master ( 7be5b2...0ef053 )
by Ramon
24s queued 10s
created

jsons._lizers_impl   A

Complexity

Total Complexity 21

Size/Duplication

Total Lines 160
Duplicated Lines 0 %

Importance

Changes 0
Metric Value
wmc 21
eloc 82
dl 0
loc 160
rs 10
c 0
b 0
f 0

7 Functions

Rating   Name   Duplication   Size   Complexity  
A set_serializer() 0 34 5
A set_deserializer() 0 35 5
A _get_lizer_by_parents() 0 12 2
A get_deserializer() 0 12 1
A _get_lizer() 0 14 3
A _get_parents() 0 17 4
A get_serializer() 0 12 1
1
"""
2
PRIVATE MODULE: do not import (from) it directly.
3
4
This module contains functionality for setting and getting serializers and
5
deserializers.
6
"""
7
from typing import Optional, Dict, Sequence, Union
8
from jsons._common_impl import StateHolder, get_class_name
9
from jsons._compatibility_impl import get_naked_class
10
11
12
def set_serializer(
13
        func: callable,
14
        cls: Union[type, Sequence[type]],
15
        high_prio: bool = True,
16
        fork_inst: type = StateHolder) -> None:
17
    """
18
    Set a serializer function for the given type. You may override the default
19
    behavior of ``jsons.load`` by setting a custom serializer.
20
21
    The ``func`` argument must take one argument (i.e. the object that is to be
22
    serialized) and also a ``kwargs`` parameter. For example:
23
24
    >>> def func(obj, **kwargs):
25
    ...    return dict()
26
27
    You may ask additional arguments between ``cls`` and ``kwargs``.
28
29
    :param func: the serializer function.
30
    :param cls: the type or sequence of types this serializer can handle.
31
    :param high_prio: determines the order in which is looked for the callable.
32
    :param fork_inst: if given, it uses this fork of ``JsonSerializable``.
33
    :return: None.
34
    """
35
    if isinstance(cls, Sequence):
36
        for cls_ in cls:
37
            set_serializer(func, cls_, high_prio, fork_inst)
38
    elif cls:
39
        index = 0 if high_prio else len(fork_inst._classes_serializers)
40
        fork_inst._classes_serializers.insert(index, cls)
41
        cls_name = get_class_name(cls, fork_inst=fork_inst,
42
                                  fully_qualified=True)
43
        fork_inst._serializers[cls_name.lower()] = func
44
    else:
45
        fork_inst._serializers['nonetype'] = func
46
47
48
def set_deserializer(
49
        func: callable,
50
        cls: Union[type, Sequence[type]],
51
        high_prio: bool = True,
52
        fork_inst: type = StateHolder) -> None:
53
    """
54
    Set a deserializer function for the given type. You may override the
55
    default behavior of ``jsons.dump`` by setting a custom deserializer.
56
57
    The ``func`` argument must take two arguments (i.e. the dict containing the
58
    serialized values and the type that the values should be deserialized into)
59
    and also a ``kwargs`` parameter. For example:
60
61
    >>> def func(dict_, cls, **kwargs):
62
    ...    return cls()
63
64
    You may ask additional arguments between ``cls`` and ``kwargs``.
65
66
    :param func: the deserializer function.
67
    :param cls: the type or sequence of types this serializer can handle.
68
    :param high_prio: determines the order in which is looked for the callable.
69
    :param fork_inst: if given, it uses this fork of ``JsonSerializable``.
70
    :return: None.
71
    """
72
    if isinstance(cls, Sequence):
73
        for cls_ in cls:
74
            set_deserializer(func, cls_, high_prio, fork_inst)
75
    elif cls:
76
        index = 0 if high_prio else len(fork_inst._classes_deserializers)
77
        fork_inst._classes_deserializers.insert(index, cls)
78
        cls_name = get_class_name(cls, fork_inst=fork_inst,
79
                                  fully_qualified=True)
80
        fork_inst._deserializers[cls_name.lower()] = func
81
    else:
82
        fork_inst._deserializers['nonetype'] = func
83
84
85
def get_serializer(
86
        cls: type,
87
        fork_inst: Optional[type] = StateHolder) -> callable:
88
    """
89
    Return the serializer function that would be used for the given ``cls``.
90
    :param cls: the type for which a serializer is to be returned.
91
    :param fork_inst: if given, it uses this fork of ``JsonSerializable``.
92
    :return: a serializer function.
93
    """
94
    serializer = _get_lizer(cls, fork_inst._serializers,
95
                            fork_inst._classes_serializers, fork_inst)
96
    return serializer
97
98
99
def get_deserializer(
100
        cls: type,
101
        fork_inst: Optional[type] = StateHolder) -> callable:
102
    """
103
    Return the deserializer function that would be used for the given ``cls``.
104
    :param cls: the type for which a deserializer is to be returned.
105
    :param fork_inst: if given, it uses this fork of ``JsonSerializable``.
106
    :return: a deserializer function.
107
    """
108
    deserializer = _get_lizer(cls, fork_inst._deserializers,
109
                              fork_inst._classes_deserializers, fork_inst)
110
    return deserializer
111
112
113
def _get_lizer(
114
        cls: type,
115
        lizers: Dict[str, callable],
116
        classes_lizers: list,
117
        fork_inst: type,
118
        recursive: bool = False) -> callable:
119
    cls_name = get_class_name(cls, str.lower, fork_inst=fork_inst,
120
                              fully_qualified=True)
121
    lizer = (lizers.get(cls_name, None)
122
             or _get_lizer_by_parents(cls, lizers, classes_lizers, fork_inst))
123
    if not lizer and not recursive:
124
        return _get_lizer(cls.__supertype__, lizers,
125
                          classes_lizers, fork_inst, True)
126
    return lizer
127
128
129
def _get_lizer_by_parents(
130
        cls: type,
131
        lizers: Dict[str, callable],
132
        classes_lizers: list,
133
        fork_inst: type) -> callable:
134
    result = None
135
    parents = _get_parents(cls, classes_lizers)
136
    if parents:
137
        pname = get_class_name(parents[0], str.lower, fork_inst=fork_inst,
138
                               fully_qualified=True)
139
        result = lizers[pname]
140
    return result
141
142
143
def _get_parents(cls: type, lizers: list) -> list:
144
    """
145
    Return a list of serializers or deserializers that can handle a parent
146
    of ``cls``.
147
    :param cls: the type that
148
    :param lizers: a list of serializers or deserializers.
149
    :return: a list of serializers or deserializers.
150
    """
151
    parents = []
152
    naked_cls = get_naked_class(cls)
153
    for cls_ in lizers:
154
        try:
155
            if issubclass(naked_cls, cls_):
156
                parents.append(cls_)
157
        except (TypeError, AttributeError):
158
            pass  # Some types do not support `issubclass` (e.g. Union).
159
    return parents
160