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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
|
from typing import Optional, TypeVar, Callable, Any, Union, cast
from dataclasses import dataclass, field, fields, replace, is_dataclass
class _DefaultType:
def __repr__(self) -> str:
return '<default>'
_DEFAULT = _DefaultType()
_ValidatorType = Callable[[str, Any], None]
_ConfigurationType = TypeVar('_ConfigurationType')
_WMIN = 2 ** 16 - 1
_4MiB = 4 * 2 ** 20
_WMAX = 2 ** 31 - 1
def _optional(validator: _ValidatorType) -> _ValidatorType:
def proc(name: str, value: Any) -> None:
if value is not None:
validator(name, value)
return proc
def _chain(*validators: _ValidatorType) -> _ValidatorType:
def proc(name: str, value: Any) -> None:
for validator in validators:
validator(name, value)
return proc
def _of_type(*types: type) -> _ValidatorType:
def proc(name: str, value: Any) -> None:
if not isinstance(value, types):
types_repr = ' or '.join(str(t) for t in types)
raise TypeError(f'"{name}" should be of type {types_repr}')
return proc
def _positive(name: str, value: Union[float, int]) -> None:
if value <= 0:
raise ValueError(f'"{name}" should be positive')
def _non_negative(name: str, value: Union[float, int]) -> None:
if value < 0:
raise ValueError(f'"{name}" should not be negative')
def _range(min_: int, max_: int) -> _ValidatorType:
def proc(name: str, value: Union[float, int]) -> None:
if value < min_:
raise ValueError(f'"{name}" should be higher or equal to {min_}')
if value > max_:
raise ValueError(f'"{name}" should be less or equal to {max_}')
return proc
def _validate(config: 'Configuration') -> None:
for f in fields(config):
validate_fn = f.metadata.get('validate')
if validate_fn is not None:
value = getattr(config, f.name)
if value is not _DEFAULT:
validate_fn(f.name, value)
def _with_defaults(
cls: _ConfigurationType, metadata_key: str,
) -> _ConfigurationType:
assert is_dataclass(cls)
defaults = {}
for f in fields(cls):
if getattr(cls, f.name) is _DEFAULT:
if metadata_key in f.metadata:
default = f.metadata[metadata_key]
else:
default = f.metadata['default']
defaults[f.name] = default
return replace(cls, **defaults) # type: ignore
@dataclass(frozen=True)
class Configuration:
_keepalive_time: Optional[float] = field(
default=cast(None, _DEFAULT),
metadata={
'validate': _optional(_chain(_of_type(int, float), _positive)),
'server-default': 7200.0,
'client-default': None,
'test-default': None,
},
)
_keepalive_timeout: float = field(
default=20.0,
metadata={
'validate': _chain(_of_type(int, float), _positive),
},
)
_keepalive_permit_without_calls: bool = field(
default=False,
metadata={
'validate': _optional(_of_type(bool)),
},
)
_http2_max_pings_without_data: int = field(
default=2,
metadata={
'validate': _optional(_chain(_of_type(int), _non_negative)),
},
)
_http2_min_sent_ping_interval_without_data: float = field(
default=300,
metadata={
'validate': _optional(_chain(_of_type(int, float), _positive)),
},
)
#: Sets inbound window size for a connection. HTTP/2 spec allows this value
#: to be from 64 KiB to 2 GiB, 4 MiB is used by default
http2_connection_window_size: int = field(
default=_4MiB,
metadata={
'validate': _chain(_of_type(int), _range(_WMIN, _WMAX)),
},
)
#: Sets inbound window size for a stream. HTTP/2 spec allows this value
#: to be from 64 KiB to 2 GiB, 4 MiB is used by default
http2_stream_window_size: int = field(
default=_4MiB,
metadata={
'validate': _chain(_of_type(int), _range(_WMIN, _WMAX)),
},
)
#: NOTE: This should be used for testing only. Overrides the hostname that
#: the target server’s certificate will be matched against. By default, the
#: value of the host argument is used.
ssl_target_name_override: Optional[str] = field(
default=None,
)
def __post_init__(self) -> None:
_validate(self)
def __for_server__(self) -> 'Configuration':
return _with_defaults(self, 'server-default')
def __for_client__(self) -> 'Configuration':
return _with_defaults(self, 'client-default')
def __for_test__(self) -> 'Configuration':
return _with_defaults(self, 'test-default')
|
