Coverage for fast_depends/pydantic/serializer.py: 100%
82 statements
« prev ^ index » next coverage.py v7.17.3, created at 2026-11-13 05:19 +0000
1import inspect 1klabmcpndqoerfghijst
2from collections.abc import Callable, Iterator, Sequence 1klabmcpndqoerfghijst
3from contextlib import contextmanager 1klabmcpndqoerfghijst
4from itertools import chain 1klabmcpndqoerfghijst
5from typing import Any 1klabmcpndqoerfghijst
7from pydantic import ValidationError as PValidationError 1klabmcpndqoerfghijst
9from fast_depends.exceptions import ValidationError 1klabmcndoefghij
10from fast_depends.library.serializer import OptionItem, Serializer, SerializerProto 1klabmcndoefghij
11from fast_depends.pydantic._compat import ( 1klabmcndoefghij
12 PYDANTIC_V2,
13 BaseModel,
14 ConfigDict,
15 PydanticUserError,
16 TypeAdapter,
17 create_model,
18 dump_json,
19 get_aliases,
20 get_config_base,
21 get_model_fields,
22)
25class PydanticSerializer(SerializerProto): 1klabmcndoefghij
26 __slots__ = ( 1klabmcndoefghij
27 "config",
28 "use_fastdepends_errors",
29 )
31 def __init__( 1klabmcndoefghij
32 self,
33 pydantic_config: ConfigDict | None = None,
34 use_fastdepends_errors: bool = True,
35 ) -> None:
36 self.config = pydantic_config 1klabmcndoefghij
37 self.use_fastdepends_errors = use_fastdepends_errors 1klabmcndoefghij
39 def __call__( 1klabmcndoefghij
40 self,
41 *,
42 name: str,
43 options: list[OptionItem],
44 response_type: Any,
45 ) -> "_PydanticSerializer":
46 if self.use_fastdepends_errors: 1klabmcndoefghij
47 if response_type is not inspect.Parameter.empty: 1klabmcndoefghij
48 return _PydanticWrappedSerializerWithResponse( 1klabmcndoefghij
49 name=name,
50 options=options,
51 response_type=response_type,
52 pydantic_config=self.config,
53 )
55 return _PydanticWrappedSerializer( 1klabmcndoefghij
56 name=name,
57 options=options,
58 pydantic_config=self.config,
59 )
61 if response_type is not inspect.Parameter.empty: 1klabmcndoefghij
62 return _PydanticSerializerWithResponse( 1klabmcndoefghij
63 name=name,
64 options=options,
65 response_type=response_type,
66 pydantic_config=self.config,
67 )
69 return _PydanticSerializer( 1klabmcndoefghij
70 name=name,
71 options=options,
72 pydantic_config=self.config,
73 )
75 @staticmethod 1klabmcndoefghij
76 def encode(message: Any) -> bytes: 1klabmcndoefghij
77 if isinstance(message, bytes): 1klabmcndoefghij
78 return message 1klabmcndoefghij
79 return dump_json(message) 1klabmcndoefghij
82class _PydanticSerializer(Serializer): 1klabmcndoefghij
83 __slots__ = ( 1klabmcndoefghij
84 "model",
85 "name",
86 "options",
87 "config",
88 "response_option",
89 )
91 def __init__( 1klabmcndoefghij
92 self,
93 *,
94 name: str,
95 options: list[OptionItem],
96 response_type: Any = None,
97 pydantic_config: ConfigDict | None = None,
98 ):
99 class_options: dict[str, Any] = { 1klabmcndoefghij
100 i.field_name: (i.field_type, i.default_value) for i in options
101 }
103 self.config = get_config_base(pydantic_config) 1klabmcndoefghij
105 self.model = create_model( # type: ignore[call-overload] 1klabmcndoefghij
106 name,
107 __config__=self.config,
108 **class_options,
109 )
111 super().__init__(name=name, options=options, response_type=response_type) 1klabmcndoefghij
113 def get_aliases(self) -> tuple[str, ...]: 1klabmcndoefghij
114 return get_aliases(self.model) 1klabmcndoefghij
116 def __call__(self, call_kwargs: dict[str, Any]) -> dict[str, Any]: 1klabmcndoefghij
117 casted_model = self.model(**call_kwargs) 1klabmcndoefghij
119 return { 1klabmcndoefghij
120 i: getattr(casted_model, i) for i in get_model_fields(casted_model).keys()
121 }
124class _PydanticSerializerWithResponse(_PydanticSerializer): 1klabmcndoefghij
125 __slots__ = ("response_callback",) 1klabmcndoefghij
127 response_callback: Callable[[Any], Any] 1klabmcndoe
129 def __init__( 1klabmcndoefghij
130 self,
131 *,
132 name: str,
133 options: list[OptionItem],
134 response_type: Any,
135 pydantic_config: ConfigDict | None = None,
136 ):
137 super().__init__( 1klabmcndoefghij
138 name=name,
139 options=options,
140 response_type=response_type,
141 pydantic_config=pydantic_config,
142 )
144 response_callback: Callable[[Any], Any] | None = None 1klabmcndoefghij
145 try: 1klabmcndoefghij
146 is_model = issubclass(response_type or object, BaseModel) 1klabmcndoefghij
147 except Exception: 1klabmcndoefghij
148 is_model = False 1klabmcndoefghij
150 if is_model: 1klabmcndoefghij
151 if PYDANTIC_V2: 1klabmcndoefghij
152 response_callback = response_type.model_validate 1abcdefghij
153 else:
154 response_callback = response_type.validate 1klmno
156 elif PYDANTIC_V2: 1klabmcndoefghij
157 try: 1abcdefghij
158 response_pydantic_type = TypeAdapter(response_type, config=self.config) 1abcdefghij
159 except PydanticUserError: 1abcdefghij
160 response_pydantic_type = TypeAdapter(response_type) 1abcdefghij
161 response_callback = response_pydantic_type.validate_python 1abcdefghij
163 if response_callback is None and not PYDANTIC_V2: 1klabmcndoefghij
164 response_model = create_model( # type: ignore[call-overload] 1klmno
165 "ResponseModel",
166 __config__=self.config,
167 r=(response_type or Any, ...),
168 )
170 def response_callback(x: Any) -> Any: 1klmno
171 return response_model(r=x).r # type: ignore[attr-defined] 1klmno
173 assert response_callback 1klabmcndoefghij
174 self.response_callback = response_callback 1klabmcndoefghij
176 def response(self, value: Any) -> Any: 1klabmcndoefghij
177 return self.response_callback(value) 1klabmcndoefghij
180class _PydanticWrappedSerializer(_PydanticSerializer): 1klabmcndoefghij
181 def __call__(self, call_kwargs: dict[str, Any]) -> dict[str, Any]: 1klabmcndoefghij
182 with self._try_pydantic(call_kwargs, self.options): 1klabmcndoefghij
183 casted_model = self.model(**call_kwargs) 1klabmcndoefghij
185 return { 1klabmcndoefghij
186 i: getattr(casted_model, i) for i in get_model_fields(casted_model).keys()
187 }
189 @contextmanager 1klabmcndoefghij
190 def _try_pydantic( 1klabmcndoefghij
191 self,
192 call_kwargs: Any,
193 options: dict[str, OptionItem],
194 locations: Sequence[Any] = (),
195 ) -> Iterator[None]:
196 try: 1klabmcndoefghij
197 yield 1klabmcndoefghij
198 except PValidationError as er: 1klabmcndoefghij
199 raise ValidationError( 1klabmcndoefghij
200 incoming_options=call_kwargs,
201 expected=options,
202 locations=locations
203 or tuple(chain(*(one_error["loc"] for one_error in er.errors()))),
204 original_error=er,
205 ) from er
208class _PydanticWrappedSerializerWithResponse( 1klabmcndoefghij
209 _PydanticWrappedSerializer,
210 _PydanticSerializerWithResponse,
211):
212 def response(self, value: Any) -> Any: 1klabmcndoefghij
213 with self._try_pydantic(value, self.response_option, ("return",)): 1klabmcndoefghij
214 return self.response_callback(value) 1klabmcndoefghij