-
Notifications
You must be signed in to change notification settings - Fork 0
/
dataclass_marshaller_factory.py
129 lines (116 loc) · 4.51 KB
/
dataclass_marshaller_factory.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
import dataclasses
import inspect
from typing import Type, Optional, List, get_type_hints, Tuple
from marshy.errors import MarshallError
from marshy.factory.marshaller_factory_abc import MarshallerFactoryABC
from marshy.marshaller import marshaller_abc
from marshy.marshaller.deferred_marshaller import DeferredMarshaller
from marshy.marshaller.obj_marshaller import ObjMarshaller, AttrConfig, attr_config
from marshy.marshaller.property_marshaller import PropertyConfig, PropertyMarshaller
from marshy.marshy_context import MarshyContext
from marshy.utils import resolve_forward_refs
@dataclasses.dataclass
class DataclassMarshallerFactory(MarshallerFactoryABC):
priority: int = 100
exclude_dumped_values: Tuple = tuple()
def create(
self, context: MarshyContext, type_: Type
) -> Optional[marshaller_abc.MarshallerABC]:
if not dataclasses.is_dataclass(type_):
return
return dataclass_marshaller(
type_, context, exclude_dumped_values=self.exclude_dumped_values
)
def get_property_configs_for_type(
type_: Type,
context: MarshyContext,
include: Optional[List[str]] = None,
exclude: Optional[List[str]] = None,
exclude_dumped_values: Tuple = DataclassMarshallerFactory.exclude_dumped_values,
):
property_configs = []
for p in inspect.getmembers(type_, lambda o: isinstance(o, property)):
name = p[0]
if skip(name, include, exclude):
continue
prop = p[1]
type_hints = get_type_hints(prop.fget)
if "return" not in type_hints:
raise MarshallError(f"UnannotatedProperty:{type_}:{name}")
prop_type = type_hints.get("return")
prop_type = resolve_forward_refs(prop_type)
marshaller = DeferredMarshaller[prop_type](prop_type, context)
property_config = PropertyConfig(
name,
marshaller,
prop,
load=bool(prop.fset),
exclude_dumped_values=exclude_dumped_values,
)
property_configs.append(property_config)
return property_configs
def skip(name: str, include: Optional[List[str]], exclude: Optional[List[str]]) -> bool:
if include is not None and name not in include:
return True
if exclude is not None and name in exclude:
return True
return False
# pylint: disable=R0913
def get_attr_configs_for_type(
type_: Type,
context: MarshyContext,
include: Optional[List[str]] = None,
exclude: Optional[List[str]] = None,
exclude_dumped_values: Tuple = (None,),
):
# noinspection PyDataclass
fields: Tuple[dataclasses.Field, ...] = dataclasses.fields(type_)
try:
types = get_type_hints(type_, globalns=None, localns=None)
attr_configs = [
attr_config(
internal_name=f.name,
marshaller=DeferredMarshaller(types[f.name], context),
exclude_dumped_values=exclude_dumped_values,
)
for f in fields
if not skip(f.name, include, exclude)
]
except NameError:
# Still supporting old style forward refs without futures...
attr_configs = [
attr_config(
internal_name=f.name,
marshaller=DeferredMarshaller(f.type, context),
exclude_dumped_values=exclude_dumped_values,
)
for f in fields
if not skip(f.name, include, exclude)
]
return attr_configs
def dataclass_marshaller(
type_: Type,
context: MarshyContext,
custom_attr_configs: Optional[List[AttrConfig]] = None,
custom_property_configs: Optional[List[PropertyConfig]] = None,
include: Optional[List[str]] = None,
exclude: Optional[List[str]] = None,
exclude_dumped_values: Tuple = (None,),
):
exclude_list = exclude or []
if custom_attr_configs:
exclude_list.extend(a.external_name for a in custom_attr_configs)
if custom_property_configs:
exclude_list.extend(p.external_name for p in custom_property_configs)
attr_configs = get_attr_configs_for_type(
type_, context, include, exclude, exclude_dumped_values
)
property_configs = get_property_configs_for_type(
type_, context, include, exclude_list, exclude_dumped_values
)
attr_configs.extend(custom_attr_configs or [])
property_configs.extend(custom_property_configs or [])
marshaller = ObjMarshaller[type_](type_, attr_configs)
if property_configs:
marshaller = PropertyMarshaller(marshaller, tuple(property_configs))
return marshaller