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
|
from collections.abc import Callable
from typing import (
Any,
Generic,
Optional,
Type,
TypeVar,
Union,
final,
overload,
)
import msgpack
from mashumaro.codecs._builder import CodecCodeBuilder
from mashumaro.core.meta.helpers import get_args
from mashumaro.dialect import Dialect
from mashumaro.mixins.msgpack import MessagePackDialect
T = TypeVar("T")
EncodedData = bytes
PostEncoderFunc = Callable[[Any], EncodedData]
PreDecoderFunc = Callable[[EncodedData], Any]
def _default_decoder(data: EncodedData) -> Any:
return msgpack.unpackb(data, raw=False)
def _default_encoder(data: Any) -> EncodedData:
return msgpack.packb(data, use_bin_type=True)
class MessagePackDecoder(Generic[T]):
@overload
def __init__(
self,
shape_type: Type[T],
*,
default_dialect: Optional[Type[Dialect]] = None,
pre_decoder_func: Optional[PreDecoderFunc] = _default_decoder,
): ...
@overload
def __init__(
self,
shape_type: Any,
*,
default_dialect: Optional[Type[Dialect]] = None,
pre_decoder_func: Optional[PreDecoderFunc] = _default_decoder,
): ...
def __init__(
self,
shape_type: Union[Type[T], Any],
*,
default_dialect: Optional[Type[Dialect]] = None,
pre_decoder_func: Optional[PreDecoderFunc] = _default_decoder,
):
if default_dialect is not None:
default_dialect = MessagePackDialect.merge(default_dialect)
else:
default_dialect = MessagePackDialect
code_builder = CodecCodeBuilder.new(
type_args=get_args(shape_type), default_dialect=default_dialect
)
code_builder.add_decode_method(shape_type, self, pre_decoder_func)
@final
def decode(self, data: EncodedData) -> T: ...
class MessagePackEncoder(Generic[T]):
@overload
def __init__(
self,
shape_type: Type[T],
*,
default_dialect: Optional[Type[Dialect]] = None,
post_encoder_func: Optional[PostEncoderFunc] = _default_encoder,
): ...
@overload
def __init__(
self,
shape_type: Any,
*,
default_dialect: Optional[Type[Dialect]] = None,
post_encoder_func: Optional[PostEncoderFunc] = _default_encoder,
): ...
def __init__(
self,
shape_type: Union[Type[T], Any],
*,
default_dialect: Optional[Type[Dialect]] = None,
post_encoder_func: Optional[PostEncoderFunc] = _default_encoder,
):
if default_dialect is not None:
default_dialect = MessagePackDialect.merge(default_dialect)
else:
default_dialect = MessagePackDialect
code_builder = CodecCodeBuilder.new(
type_args=get_args(shape_type), default_dialect=default_dialect
)
code_builder.add_encode_method(shape_type, self, post_encoder_func)
@final
def encode(self, obj: T) -> EncodedData: ...
def msgpack_decode(data: EncodedData, shape_type: Union[Type[T], Any]) -> T:
return MessagePackDecoder(shape_type).decode(data)
def msgpack_encode(obj: T, shape_type: Union[Type[T], Any]) -> EncodedData:
return MessagePackEncoder(shape_type).encode(obj)
decode = msgpack_decode
encode = msgpack_encode
__all__ = [
"MessagePackDecoder",
"MessagePackEncoder",
"msgpack_decode",
"msgpack_encode",
"decode",
"encode",
]
|
