/
__init__.py
138 lines (112 loc) · 5.12 KB
/
__init__.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
130
131
132
133
134
135
136
137
138
import dataclasses
from enum import Enum
from inspect import isclass
from typing import Any, Dict, Mapping, Optional, Union
import log
from ruamel.yaml.scalarfloat import ScalarFloat
from ..utils import cached, subclasses
from ._bases import Converter
from .builtins import Boolean, Float, Integer, String
from .containers import Dataclass, Dictionary, List, Set
from .enumerations import Enumeration
from .extensions import * # pylint: disable=unused-wildcard-import
_REGISTRY: Dict[Union[type, str], type] = {}
def register(cls: Union[type, str], converter: type):
"""Associate the given type signature with a converter class."""
_REGISTRY[cls] = converter
if not isinstance(cls, str):
_REGISTRY[cls.__name__] = converter
register(Integer.TYPE, Integer)
register(Float.TYPE, Float)
register(ScalarFloat, Float)
register(Boolean.TYPE, Boolean)
register(String.TYPE, String)
@cached
def map_type(cls, *, name: str = '', item_cls: Optional[type] = None):
"""Infer the converter type from a dataclass, type, or annotation."""
if name:
log.debug(f'Mapping {name!r} of {cls!r} to converter')
else:
log.debug(f'Mapping {cls!r} to converter')
if cls in _REGISTRY:
converter: Any = _REGISTRY[cls]
log.debug(f'Mapped {cls!r} to existing converter: {converter}')
return converter
if dataclasses.is_dataclass(cls):
converters = {}
for field in dataclasses.fields(cls):
converters[field.name] = map_type(field.type, name=field.name) # type: ignore
converter = Dataclass.of_mappings(cls, converters)
log.debug(f'Mapped {cls!r} to new converter: {converter}')
return converter
if hasattr(cls, '__origin__'):
converter = None
if cls.__origin__ == list:
try:
converter = map_type(item_cls or cls.__args__[0])
except TypeError as e: # Python 3.8 behavior
assert '~T' in str(e), f'Unhandled error: {e}'
raise TypeError("Type is required with 'List' annotation") from None
except AttributeError as e: # Python 3.9 behavior
assert '__args__' in str(e), f'Unhandled error: {e}'
raise TypeError("Type is required with 'List' annotation") from None
else:
converter = List.of_type(converter)
elif cls.__origin__ == set:
try:
converter = map_type(item_cls or cls.__args__[0])
except TypeError as e: # Python 3.8 behavior
assert '~T' in str(e), f'Unhandled error: {e}'
raise TypeError("Type is required with 'Set' annotation") from None
except AttributeError as e: # Python 3.9 behavior
assert '__args__' in str(e), f'Unhandled error: {e}'
raise TypeError("Type is required with 'Set' annotation") from None
else:
converter = Set.of_type(converter)
elif isclass(cls.__origin__) and issubclass(cls.__origin__, Mapping):
if item_cls:
key = map_type(str)
value = map_type(item_cls)
else:
log.warn("Schema enforcement not possible with 'Dict' annotation")
try:
key = map_type(cls.__args__[0])
value = map_type(cls.__args__[1])
except TypeError as e: # Python 3.8 behavior
assert '~' in str(e), f'Unhandled error: {e}'
raise TypeError(
"Types are required with 'Dict' annotation"
) from None
except AttributeError as e: # Python 3.9 behavior
assert '__args__' in str(e), f'Unhandled error: {e}'
raise TypeError(
"Types are required with 'Dict' annotation"
) from None
converter = Dictionary.of_mapping(key, value)
elif cls.__origin__ == Union:
converter = map_type(cls.__args__[0])
assert len(cls.__args__) == 2
assert cls.__args__[1] == type(None)
converter = converter.as_optional()
elif issubclass(cls.__origin__, Converter):
subtypes = [map_type(t) for t in cls.__args__]
converter = cls.__origin__.as_generic(subtypes)
if converter:
log.debug(f'Mapped {cls!r} to new converter: {converter}')
return converter
raise TypeError(f'Unsupported container type: {cls.__origin__}')
if isinstance(cls, str):
log.debug(f'Searching for class matching {cls!r} annotation')
for cls2 in subclasses(Converter):
if cls2.__name__ == cls:
register(cls, cls2)
log.debug(f'Registered {cls2} as new converter')
return cls2
if not isclass(cls):
raise TypeError(f'Annotation is not a type: {cls!r}')
if issubclass(cls, Converter):
log.debug(f'Mapped {cls!r} to existing converter (itself)')
return cls
if issubclass(cls, Enum):
return Enumeration.of_type(cls)
raise TypeError(f'Could not map type: {cls}')