Coverage for fast_depends/msgspec/serializer.py: 100%
73 statements
« prev ^ index » next coverage.py v7.17.3, created at 2026-11-13 05:19 +0000
1import inspect 1ijklmnabcdefghopqrst
2import re 1ijklmnabcdefghopqrst
3from collections.abc import Callable, Iterator, Sequence 1ijklmnabcdefghopqrst
4from contextlib import contextmanager 1ijklmnabcdefghopqrst
5from typing import Any, TypeVar 1ijklmnabcdefghopqrst
7import msgspec 1ijklmnabcdefghopqrst
9from fast_depends.exceptions import ValidationError 1abcdefgh
10from fast_depends.library.serializer import OptionItem, Serializer, SerializerProto 1abcdefgh
12T = TypeVar("T") 1abcdefgh
15class MsgSpecSerializer(SerializerProto): 1abcdefgh
16 __slots__ = ("use_fastdepends_errors", "dec_hook") 1abcdefgh
18 def __init__( 1abcdefgh
19 self,
20 use_fastdepends_errors: bool = True,
21 dec_hook: Callable[[type[T], Any], T] | None = None,
22 ) -> None:
23 self.use_fastdepends_errors = use_fastdepends_errors 1abcdefgh
24 self.dec_hook = dec_hook 1abcdefgh
26 def __call__( 1abcdefgh
27 self,
28 *,
29 name: str,
30 options: list[OptionItem],
31 response_type: Any,
32 ) -> "_MsgSpecSerializer":
33 if self.use_fastdepends_errors: 1abcdefgh
34 if response_type is not inspect.Parameter.empty: 1abcdefgh
35 return _MsgSpecWrappedSerializerWithResponse( 1abcdefgh
36 name=name,
37 options=options,
38 response_type=response_type,
39 dec_hook=self.dec_hook,
40 )
42 return _MsgSpecWrappedSerializer( 1abcdefgh
43 name=name,
44 options=options,
45 dec_hook=self.dec_hook,
46 )
48 if response_type is not inspect.Parameter.empty: 1abcdefgh
49 return _MsgSpecSerializerWithResponse( 1abcdefgh
50 name=name,
51 options=options,
52 response_type=response_type,
53 dec_hook=self.dec_hook,
54 )
56 return _MsgSpecSerializer( 1abcdefgh
57 name=name,
58 options=options,
59 dec_hook=self.dec_hook,
60 )
62 @staticmethod 1abcdefgh
63 def encode(message: Any) -> bytes: 1abcdefgh
64 if isinstance(message, bytes): 1abcdefgh
65 return message 1abcdefgh
66 return msgspec.json.encode(message) 1abcdefgh
69class _MsgSpecSerializer(Serializer): 1abcdefgh
70 __slots__ = ( 1abcdefgh
71 "aliases",
72 "model",
73 "response_type",
74 "name",
75 "options",
76 "response_option",
77 "dec_hook",
78 )
80 def __init__( 1abcdefgh
81 self,
82 *,
83 name: str,
84 options: list[OptionItem],
85 response_type: Any = None,
86 dec_hook: Callable[[type[T], Any], T] | None = None,
87 ):
88 model_options: list[str | tuple[str, type] | tuple[str, type, Any]] = [] 1abcdefgh
89 aliases = {} 1abcdefgh
90 for i in options: 1abcdefgh
91 default_value = i.default_value 1abcdefgh
93 if isinstance(default_value, msgspec._core.Field) and default_value.name: 1abcdefgh
94 aliases[i.field_name] = default_value.name 1abcdefgh
95 else:
96 aliases[i.field_name] = i.field_name 1abcdefgh
98 if default_value is Ellipsis: 1abcdefgh
99 model_options.append( 1abcdefgh
100 (
101 i.field_name,
102 i.field_type,
103 )
104 )
105 else:
106 model_options.append( 1abcdefgh
107 (
108 i.field_name,
109 i.field_type,
110 default_value,
111 )
112 )
114 self.aliases = aliases 1abcdefgh
115 self.model = msgspec.defstruct(name, model_options, kw_only=True) 1abcdefgh
116 self.dec_hook = dec_hook 1abcdefgh
117 super().__init__(name=name, options=options, response_type=response_type) 1abcdefgh
119 def get_aliases(self) -> tuple[str, ...]: 1abcdefgh
120 return tuple(self.aliases.values()) 1abcdefgh
122 def __call__(self, call_kwargs: dict[str, Any]) -> dict[str, Any]: 1abcdefgh
123 casted_model = msgspec.convert( 1abcdefgh
124 call_kwargs,
125 type=self.model,
126 strict=False,
127 str_keys=True,
128 dec_hook=self.dec_hook,
129 )
131 return { 1abcdefgh
132 out_field: getattr(casted_model, out_field, None)
133 for out_field in self.aliases.keys()
134 }
137class _MsgSpecSerializerWithResponse(_MsgSpecSerializer): 1abcdefgh
138 def __init__( 1abcdefgh
139 self,
140 *,
141 name: str,
142 options: list[OptionItem],
143 response_type: Any,
144 dec_hook: Callable[[type[T], Any], T] | None = None,
145 ):
146 super().__init__( 1abcdefgh
147 name=name,
148 options=options,
149 response_type=response_type,
150 dec_hook=dec_hook,
151 )
152 self.response_type = response_type 1abcdefgh
154 def response(self, value: Any) -> Any: 1abcdefgh
155 return msgspec.convert( 1abcdefgh
156 value,
157 type=self.response_type,
158 strict=False,
159 dec_hook=self.dec_hook,
160 )
163class _MsgSpecWrappedSerializer(_MsgSpecSerializer): 1abcdefgh
164 def __call__(self, call_kwargs: dict[str, Any]) -> dict[str, Any]: 1abcdefgh
165 with self._try_msgspec(call_kwargs, self.options): 1abcdefgh
166 casted_model = msgspec.convert( 1abcdefgh
167 call_kwargs,
168 type=self.model,
169 strict=False,
170 str_keys=True,
171 dec_hook=self.dec_hook,
172 )
174 return { 1abcdefgh
175 out_field: getattr(casted_model, out_field, None)
176 for out_field in self.aliases.keys()
177 }
179 @contextmanager 1abcdefgh
180 def _try_msgspec( 1abcdefgh
181 self,
182 call_kwargs: Any,
183 options: dict[str, OptionItem],
184 locations: Sequence[str] = (),
185 ) -> Iterator[None]:
186 try: 1abcdefgh
187 yield 1abcdefgh
188 except msgspec.ValidationError as er: 1abcdefgh
189 raise ValidationError( 1abcdefgh
190 incoming_options=call_kwargs,
191 expected=options,
192 locations=locations or re.findall(r"at `\$\.(.)`", str(er.args)),
193 original_error=er,
194 ) from er
197class _MsgSpecWrappedSerializerWithResponse(_MsgSpecWrappedSerializer): 1abcdefgh
198 def __init__( 1abcdefgh
199 self,
200 *,
201 name: str,
202 options: list[OptionItem],
203 response_type: Any,
204 dec_hook: Callable[[type[T], Any], T] | None = None,
205 ):
206 super().__init__( 1abcdefgh
207 name=name,
208 options=options,
209 response_type=response_type,
210 dec_hook=dec_hook,
211 )
212 self.response_type = response_type 1abcdefgh
214 def response(self, value: Any) -> Any: 1abcdefgh
215 with self._try_msgspec(value, self.response_option, ("return",)): 1abcdefgh
216 return msgspec.convert( 1abcdefgh
217 value,
218 type=self.response_type,
219 strict=False,
220 dec_hook=self.dec_hook,
221 )