Coverage for pydantic/functional_serializers.py: 95.65%
65 statements
« prev ^ index » next coverage.py v7.6.12, created at 2025-02-13 19:35 +0000
1"""This module contains related classes and functions for serialization."""
3from __future__ import annotations 1opqrstabcdmnuvwxyzefghABCDEFijkl
5import dataclasses 1opqrstabcdmnuvwxyzefghABCDEFijkl
6from functools import partial, partialmethod 1opqrstabcdmnuvwxyzefghABCDEFijkl
7from typing import TYPE_CHECKING, Annotated, Any, Callable, Literal, TypeVar, overload 1opqrstabcdmnuvwxyzefghABCDEFijkl
9from pydantic_core import PydanticUndefined, core_schema 1opqrstabcdmnuvwxyzefghABCDEFijkl
10from pydantic_core.core_schema import SerializationInfo, SerializerFunctionWrapHandler, WhenUsed 1opqrstabcdmnuvwxyzefghABCDEFijkl
11from typing_extensions import TypeAlias 1opqrstabcdmnuvwxyzefghABCDEFijkl
13from . import PydanticUndefinedAnnotation 1opqrstabcdmnuvwxyzefghABCDEFijkl
14from ._internal import _decorators, _internal_dataclass 1opqrstabcdmnuvwxyzefghABCDEFijkl
15from .annotated_handlers import GetCoreSchemaHandler 1opqrstabcdmnuvwxyzefghABCDEFijkl
18@dataclasses.dataclass(**_internal_dataclass.slots_true, frozen=True) 1opqrstabcdmnuvwxyzefghABCDEFijkl
19class PlainSerializer: 1opqrstabcdmnuvwxyzefghABCDEFijkl
20 """Plain serializers use a function to modify the output of serialization.
22 This is particularly helpful when you want to customize the serialization for annotated types.
23 Consider an input of `list`, which will be serialized into a space-delimited string.
25 ```python
26 from typing import Annotated
28 from pydantic import BaseModel, PlainSerializer
30 CustomStr = Annotated[
31 list, PlainSerializer(lambda x: ' '.join(x), return_type=str)
32 ]
34 class StudentModel(BaseModel):
35 courses: CustomStr
37 student = StudentModel(courses=['Math', 'Chemistry', 'English'])
38 print(student.model_dump())
39 #> {'courses': 'Math Chemistry English'}
40 ```
42 Attributes:
43 func: The serializer function.
44 return_type: The return type for the function. If omitted it will be inferred from the type annotation.
45 when_used: Determines when this serializer should be used. Accepts a string with values `'always'`,
46 `'unless-none'`, `'json'`, and `'json-unless-none'`. Defaults to 'always'.
47 """
49 func: core_schema.SerializerFunction 1opqrstabcdmnuvwxyzefghABCDEFijkl
50 return_type: Any = PydanticUndefined 1opqrstabcdmnuvwxyzefghABCDEFijkl
51 when_used: WhenUsed = 'always' 1opqrstabcdmnuvwxyzefghABCDEFijkl
53 def __get_pydantic_core_schema__(self, source_type: Any, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema: 1opqrstabcdmnuvwxyzefghABCDEFijkl
54 """Gets the Pydantic core schema.
56 Args:
57 source_type: The source type.
58 handler: The `GetCoreSchemaHandler` instance.
60 Returns:
61 The Pydantic core schema.
62 """
63 schema = handler(source_type) 1opqrstabcdmnuvwxyzefghABCDEFijkl
64 try: 1opqrstabcdmnuvwxyzefghABCDEFijkl
65 # Do not pass in globals as the function could be defined in a different module.
66 # Instead, let `get_function_return_type` infer the globals to use, but still pass
67 # in locals that may contain a parent/rebuild namespace:
68 return_type = _decorators.get_function_return_type( 1opqrstabcdmnuvwxyzefghABCDEFijkl
69 self.func,
70 self.return_type,
71 localns=handler._get_types_namespace().locals,
72 )
73 except NameError as e: 1opqrstabcdmnuvwxyzefghABCDEFijkl
74 raise PydanticUndefinedAnnotation.from_name_error(e) from e 1opqrstabcdmnuvwxyzefghABCDEFijkl
75 return_schema = None if return_type is PydanticUndefined else handler.generate_schema(return_type) 1opqrstabcdmnuvwxyzefghABCDEFijkl
76 schema['serialization'] = core_schema.plain_serializer_function_ser_schema( 1opqrstabcdmnuvwxyzefghABCDEFijkl
77 function=self.func,
78 info_arg=_decorators.inspect_annotated_serializer(self.func, 'plain'),
79 return_schema=return_schema,
80 when_used=self.when_used,
81 )
82 return schema 1opqrstabcdmnuvwxyzefghABCDEFijkl
85@dataclasses.dataclass(**_internal_dataclass.slots_true, frozen=True) 1opqrstabcdmnuvwxyzefghABCDEFijkl
86class WrapSerializer: 1opqrstabcdmnuvwxyzefghABCDEFijkl
87 """Wrap serializers receive the raw inputs along with a handler function that applies the standard serialization
88 logic, and can modify the resulting value before returning it as the final output of serialization.
90 For example, here's a scenario in which a wrap serializer transforms timezones to UTC **and** utilizes the existing `datetime` serialization logic.
92 ```python
93 from datetime import datetime, timezone
94 from typing import Annotated, Any
96 from pydantic import BaseModel, WrapSerializer
98 class EventDatetime(BaseModel):
99 start: datetime
100 end: datetime
102 def convert_to_utc(value: Any, handler, info) -> dict[str, datetime]:
103 # Note that `handler` can actually help serialize the `value` for
104 # further custom serialization in case it's a subclass.
105 partial_result = handler(value, info)
106 if info.mode == 'json':
107 return {
108 k: datetime.fromisoformat(v).astimezone(timezone.utc)
109 for k, v in partial_result.items()
110 }
111 return {k: v.astimezone(timezone.utc) for k, v in partial_result.items()}
113 UTCEventDatetime = Annotated[EventDatetime, WrapSerializer(convert_to_utc)]
115 class EventModel(BaseModel):
116 event_datetime: UTCEventDatetime
118 dt = EventDatetime(
119 start='2024-01-01T07:00:00-08:00', end='2024-01-03T20:00:00+06:00'
120 )
121 event = EventModel(event_datetime=dt)
122 print(event.model_dump())
123 '''
124 {
125 'event_datetime': {
126 'start': datetime.datetime(
127 2024, 1, 1, 15, 0, tzinfo=datetime.timezone.utc
128 ),
129 'end': datetime.datetime(
130 2024, 1, 3, 14, 0, tzinfo=datetime.timezone.utc
131 ),
132 }
133 }
134 '''
136 print(event.model_dump_json())
137 '''
138 {"event_datetime":{"start":"2024-01-01T15:00:00Z","end":"2024-01-03T14:00:00Z"}}
139 '''
140 ```
142 Attributes:
143 func: The serializer function to be wrapped.
144 return_type: The return type for the function. If omitted it will be inferred from the type annotation.
145 when_used: Determines when this serializer should be used. Accepts a string with values `'always'`,
146 `'unless-none'`, `'json'`, and `'json-unless-none'`. Defaults to 'always'.
147 """
149 func: core_schema.WrapSerializerFunction 1opqrstabcdmnuvwxyzefghABCDEFijkl
150 return_type: Any = PydanticUndefined 1opqrstabcdmnuvwxyzefghABCDEFijkl
151 when_used: WhenUsed = 'always' 1opqrstabcdmnuvwxyzefghABCDEFijkl
153 def __get_pydantic_core_schema__(self, source_type: Any, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema: 1opqrstabcdmnuvwxyzefghABCDEFijkl
154 """This method is used to get the Pydantic core schema of the class.
156 Args:
157 source_type: Source type.
158 handler: Core schema handler.
160 Returns:
161 The generated core schema of the class.
162 """
163 schema = handler(source_type) 1opqrstabcdmnuvwxyzefghABCDEFijkl
164 globalns, localns = handler._get_types_namespace() 1opqrstabcdmnuvwxyzefghABCDEFijkl
165 try: 1opqrstabcdmnuvwxyzefghABCDEFijkl
166 # Do not pass in globals as the function could be defined in a different module.
167 # Instead, let `get_function_return_type` infer the globals to use, but still pass
168 # in locals that may contain a parent/rebuild namespace:
169 return_type = _decorators.get_function_return_type( 1opqrstabcdmnuvwxyzefghABCDEFijkl
170 self.func,
171 self.return_type,
172 localns=handler._get_types_namespace().locals,
173 )
174 except NameError as e: 1opqrstabcdmnuvwxyzefghABCDEFijkl
175 raise PydanticUndefinedAnnotation.from_name_error(e) from e 1opqrstabcdmnuvwxyzefghABCDEFijkl
176 return_schema = None if return_type is PydanticUndefined else handler.generate_schema(return_type) 1opqrstabcdmnuvwxyzefghABCDEFijkl
177 schema['serialization'] = core_schema.wrap_serializer_function_ser_schema( 1opqrstabcdmnuvwxyzefghABCDEFijkl
178 function=self.func,
179 info_arg=_decorators.inspect_annotated_serializer(self.func, 'wrap'),
180 return_schema=return_schema,
181 when_used=self.when_used,
182 )
183 return schema 1opqrstabcdmnuvwxyzefghABCDEFijkl
186if TYPE_CHECKING: 1opqrstabcdmnuvwxyzefghABCDEFijkl
187 _Partial: TypeAlias = 'partial[Any] | partialmethod[Any]'
189 FieldPlainSerializer: TypeAlias = 'core_schema.SerializerFunction | _Partial'
190 """A field serializer method or function in `plain` mode."""
192 FieldWrapSerializer: TypeAlias = 'core_schema.WrapSerializerFunction | _Partial'
193 """A field serializer method or function in `wrap` mode."""
195 FieldSerializer: TypeAlias = 'FieldPlainSerializer | FieldWrapSerializer'
196 """A field serializer method or function."""
198 _FieldPlainSerializerT = TypeVar('_FieldPlainSerializerT', bound=FieldPlainSerializer)
199 _FieldWrapSerializerT = TypeVar('_FieldWrapSerializerT', bound=FieldWrapSerializer)
202@overload 1opqrstabcdmnuvwxyzefghABCDEFijkl
203def field_serializer( 1opqrstabcdmnuvwxyzefghABCDEFijkl
204 field: str, 1abcdmnefghijkl
205 /,
206 *fields: str, 1abcdmnefghijkl
207 mode: Literal['wrap'], 1abcdmnefghijkl
208 return_type: Any = ..., 1opqrstabcdmnuvwxyzefghABCDEFijkl
209 when_used: WhenUsed = ..., 1opqrstabcdmnuvwxyzefghABCDEFijkl
210 check_fields: bool | None = ..., 1opqrstabcdmnuvwxyzefghABCDEFijkl
211) -> Callable[[_FieldWrapSerializerT], _FieldWrapSerializerT]: ... 1abcdmnefghijkl
214@overload 1opqrstabcdmnuvwxyzefghABCDEFijkl
215def field_serializer( 1opqrstabcdmnuvwxyzefghABCDEFijkl
216 field: str, 1abcdmnefghijkl
217 /,
218 *fields: str, 1abcdmnefghijkl
219 mode: Literal['plain'] = ..., 1opqrstabcdmnuvwxyzefghABCDEFijkl
220 return_type: Any = ..., 1opqrstabcdmnuvwxyzefghABCDEFijkl
221 when_used: WhenUsed = ..., 1opqrstabcdmnuvwxyzefghABCDEFijkl
222 check_fields: bool | None = ..., 1opqrstabcdmnuvwxyzefghABCDEFijkl
223) -> Callable[[_FieldPlainSerializerT], _FieldPlainSerializerT]: ... 1abcdmnefghijkl
226def field_serializer( 1opqrstabcdmnuvwxyzefghABCDEFijkl
227 *fields: str,
228 mode: Literal['plain', 'wrap'] = 'plain',
229 return_type: Any = PydanticUndefined,
230 when_used: WhenUsed = 'always',
231 check_fields: bool | None = None,
232) -> (
233 Callable[[_FieldWrapSerializerT], _FieldWrapSerializerT]
234 | Callable[[_FieldPlainSerializerT], _FieldPlainSerializerT]
235):
236 """Decorator that enables custom field serialization.
238 In the below example, a field of type `set` is used to mitigate duplication. A `field_serializer` is used to serialize the data as a sorted list.
240 ```python
241 from typing import Set
243 from pydantic import BaseModel, field_serializer
245 class StudentModel(BaseModel):
246 name: str = 'Jane'
247 courses: Set[str]
249 @field_serializer('courses', when_used='json')
250 def serialize_courses_in_order(self, courses: Set[str]):
251 return sorted(courses)
253 student = StudentModel(courses={'Math', 'Chemistry', 'English'})
254 print(student.model_dump_json())
255 #> {"name":"Jane","courses":["Chemistry","English","Math"]}
256 ```
258 See [Custom serializers](../concepts/serialization.md#custom-serializers) for more information.
260 Four signatures are supported:
262 - `(self, value: Any, info: FieldSerializationInfo)`
263 - `(self, value: Any, nxt: SerializerFunctionWrapHandler, info: FieldSerializationInfo)`
264 - `(value: Any, info: SerializationInfo)`
265 - `(value: Any, nxt: SerializerFunctionWrapHandler, info: SerializationInfo)`
267 Args:
268 fields: Which field(s) the method should be called on.
269 mode: The serialization mode.
271 - `plain` means the function will be called instead of the default serialization logic,
272 - `wrap` means the function will be called with an argument to optionally call the
273 default serialization logic.
274 return_type: Optional return type for the function, if omitted it will be inferred from the type annotation.
275 when_used: Determines the serializer will be used for serialization.
276 check_fields: Whether to check that the fields actually exist on the model.
278 Returns:
279 The decorator function.
280 """
282 def dec(f: FieldSerializer) -> _decorators.PydanticDescriptorProxy[Any]: 1opqrstabcdmnuvwxyzefghABCDEFijkl
283 dec_info = _decorators.FieldSerializerDecoratorInfo( 1opqrstabcdmnuvwxyzefghABCDEFijkl
284 fields=fields,
285 mode=mode,
286 return_type=return_type,
287 when_used=when_used,
288 check_fields=check_fields,
289 )
290 return _decorators.PydanticDescriptorProxy(f, dec_info) # pyright: ignore[reportArgumentType] 1opqrstabcdmnuvwxyzefghABCDEFijkl
292 return dec # pyright: ignore[reportReturnType] 1opqrstabcdmnuvwxyzefghABCDEFijkl
295if TYPE_CHECKING: 1opqrstabcdmnuvwxyzefghABCDEFijkl
296 # The first argument in the following callables represent the `self` type:
298 ModelPlainSerializerWithInfo: TypeAlias = Callable[[Any, SerializationInfo], Any]
299 """A model serializer method with the `info` argument, in `plain` mode."""
301 ModelPlainSerializerWithoutInfo: TypeAlias = Callable[[Any], Any]
302 """A model serializer method without the `info` argument, in `plain` mode."""
304 ModelPlainSerializer: TypeAlias = 'ModelPlainSerializerWithInfo | ModelPlainSerializerWithoutInfo'
305 """A model serializer method in `plain` mode."""
307 ModelWrapSerializerWithInfo: TypeAlias = Callable[[Any, SerializerFunctionWrapHandler, SerializationInfo], Any]
308 """A model serializer method with the `info` argument, in `wrap` mode."""
310 ModelWrapSerializerWithoutInfo: TypeAlias = Callable[[Any, SerializerFunctionWrapHandler], Any]
311 """A model serializer method without the `info` argument, in `wrap` mode."""
313 ModelWrapSerializer: TypeAlias = 'ModelWrapSerializerWithInfo | ModelWrapSerializerWithoutInfo'
314 """A model serializer method in `wrap` mode."""
316 ModelSerializer: TypeAlias = 'ModelPlainSerializer | ModelWrapSerializer'
318 _ModelPlainSerializerT = TypeVar('_ModelPlainSerializerT', bound=ModelPlainSerializer)
319 _ModelWrapSerializerT = TypeVar('_ModelWrapSerializerT', bound=ModelWrapSerializer)
322@overload 1opqrstabcdmnuvwxyzefghABCDEFijkl
323def model_serializer(f: _ModelPlainSerializerT, /) -> _ModelPlainSerializerT: ... 1opqrstabcdmnuvwxyzefghABCDEFijkl
326@overload 1opqrstabcdmnuvwxyzefghABCDEFijkl
327def model_serializer( 1opqrstabcdmnuvwxyzefghABCDEFijkl
328 *, mode: Literal['wrap'], when_used: WhenUsed = 'always', return_type: Any = ... 1opqrstabcdmnuvwxyzefghABCDEFijkl
329) -> Callable[[_ModelWrapSerializerT], _ModelWrapSerializerT]: ... 1abcdmnefghijkl
332@overload 1opqrstabcdmnuvwxyzefghABCDEFijkl
333def model_serializer( 1opqrstabcdmnuvwxyzefghABCDEFijkl
334 *,
335 mode: Literal['plain'] = ..., 1opqrstabcdmnuvwxyzefghABCDEFijkl
336 when_used: WhenUsed = 'always', 1opqrstabcdmnuvwxyzefghABCDEFijkl
337 return_type: Any = ..., 1opqrstabcdmnuvwxyzefghABCDEFijkl
338) -> Callable[[_ModelPlainSerializerT], _ModelPlainSerializerT]: ... 1abcdmnefghijkl
341def model_serializer( 1opqrstabcduvwxyzefghABCDEFijkl
342 f: _ModelPlainSerializerT | _ModelWrapSerializerT | None = None,
343 /,
344 *,
345 mode: Literal['plain', 'wrap'] = 'plain',
346 when_used: WhenUsed = 'always',
347 return_type: Any = PydanticUndefined,
348) -> (
349 _ModelPlainSerializerT
350 | Callable[[_ModelWrapSerializerT], _ModelWrapSerializerT]
351 | Callable[[_ModelPlainSerializerT], _ModelPlainSerializerT]
352):
353 """Decorator that enables custom model serialization.
355 This is useful when a model need to be serialized in a customized manner, allowing for flexibility beyond just specific fields.
357 An example would be to serialize temperature to the same temperature scale, such as degrees Celsius.
359 ```python
360 from typing import Literal
362 from pydantic import BaseModel, model_serializer
364 class TemperatureModel(BaseModel):
365 unit: Literal['C', 'F']
366 value: int
368 @model_serializer()
369 def serialize_model(self):
370 if self.unit == 'F':
371 return {'unit': 'C', 'value': int((self.value - 32) / 1.8)}
372 return {'unit': self.unit, 'value': self.value}
374 temperature = TemperatureModel(unit='F', value=212)
375 print(temperature.model_dump())
376 #> {'unit': 'C', 'value': 100}
377 ```
379 Two signatures are supported for `mode='plain'`, which is the default:
381 - `(self)`
382 - `(self, info: SerializationInfo)`
384 And two other signatures for `mode='wrap'`:
386 - `(self, nxt: SerializerFunctionWrapHandler)`
387 - `(self, nxt: SerializerFunctionWrapHandler, info: SerializationInfo)`
389 See [Custom serializers](../concepts/serialization.md#custom-serializers) for more information.
391 Args:
392 f: The function to be decorated.
393 mode: The serialization mode.
395 - `'plain'` means the function will be called instead of the default serialization logic
396 - `'wrap'` means the function will be called with an argument to optionally call the default
397 serialization logic.
398 when_used: Determines when this serializer should be used.
399 return_type: The return type for the function. If omitted it will be inferred from the type annotation.
401 Returns:
402 The decorator function.
403 """
405 def dec(f: ModelSerializer) -> _decorators.PydanticDescriptorProxy[Any]: 1opqrstabcdmnuvwxyzefghABCDEFijkl
406 dec_info = _decorators.ModelSerializerDecoratorInfo(mode=mode, return_type=return_type, when_used=when_used) 1opqrstabcdmnuvwxyzefghABCDEFijkl
407 return _decorators.PydanticDescriptorProxy(f, dec_info) 1opqrstabcdmnuvwxyzefghABCDEFijkl
409 if f is None: 1opqrstabcdmnuvwxyzefghABCDEFijkl
410 return dec # pyright: ignore[reportReturnType] 1opqrstabcdmnuvwxyzefghABCDEFijkl
411 else:
412 return dec(f) # pyright: ignore[reportReturnType] 1opqrstabcdmnuvwxyzefghABCDEFijkl
415AnyType = TypeVar('AnyType') 1opqrstabcdmnuvwxyzefghABCDEFijkl
418if TYPE_CHECKING: 1opqrstabcdmnuvwxyzefghABCDEFijkl
419 SerializeAsAny = Annotated[AnyType, ...] # SerializeAsAny[list[str]] will be treated by type checkers as list[str]
420 """Force serialization to ignore whatever is defined in the schema and instead ask the object
421 itself how it should be serialized.
422 In particular, this means that when model subclasses are serialized, fields present in the subclass
423 but not in the original schema will be included.
424 """
425else:
427 @dataclasses.dataclass(**_internal_dataclass.slots_true) 1opqrstabcdmnuvwxyzefghABCDEFijkl
428 class SerializeAsAny: # noqa: D101 1opqrstabcdmnuvwxyzefghABCDEFijkl
429 def __class_getitem__(cls, item: Any) -> Any: 1opqrstabcdmnuvwxyzefghABCDEFijkl
430 return Annotated[item, SerializeAsAny()] 1opqrstabcdmnuvwxyzefghABCDEFijkl
432 def __get_pydantic_core_schema__( 1opqrstabcdmnuvwxyzefghABCDEFijkl
433 self, source_type: Any, handler: GetCoreSchemaHandler
434 ) -> core_schema.CoreSchema:
435 schema = handler(source_type) 1opqrstabcdmnuvwxyzefghABCDEFijkl
436 schema_to_update = schema 1opqrstabcdmnuvwxyzefghABCDEFijkl
437 while schema_to_update['type'] == 'definitions': 437 ↛ 438line 437 didn't jump to line 438 because the condition on line 437 was never true1opqrstabcdmnuvwxyzefghABCDEFijkl
438 schema_to_update = schema_to_update.copy()
439 schema_to_update = schema_to_update['schema']
440 schema_to_update['serialization'] = core_schema.wrap_serializer_function_ser_schema( 1opqrstabcdmnuvwxyzefghABCDEFijkl
441 lambda x, h: h(x), schema=core_schema.any_schema()
442 )
443 return schema 1opqrstabcdmnuvwxyzefghABCDEFijkl
445 __hash__ = object.__hash__ 1opqrstabcdmnuvwxyzefghABCDEFijkl