Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""This module contains related classes and functions for serialization."""
3from __future__ import annotations
5import dataclasses
6from collections.abc import Callable
7from functools import partial, partialmethod
8from typing import TYPE_CHECKING, Annotated, Any, Literal, TypeAlias, TypeVar, overload
10from pydantic_core import PydanticUndefined, core_schema
11from pydantic_core.core_schema import SerializationInfo, SerializerFunctionWrapHandler, WhenUsed
12from typing_extensions import TypeForm
14from . import PydanticUndefinedAnnotation
15from ._internal import _decorators
16from .annotated_handlers import GetCoreSchemaHandler
17from .errors import PydanticUserError
20@dataclasses.dataclass(slots=True, frozen=True)
21class PlainSerializer:
22 """Plain serializers use a function to modify the output of serialization.
24 This is particularly helpful when you want to customize the serialization for annotated types.
25 Consider an input of `list`, which will be serialized into a space-delimited string.
27 ```python
28 from typing import Annotated
30 from pydantic import BaseModel, PlainSerializer
32 CustomStr = Annotated[
33 list, PlainSerializer(lambda x: ' '.join(x), return_type=str)
34 ]
36 class StudentModel(BaseModel):
37 courses: CustomStr
39 student = StudentModel(courses=['Math', 'Chemistry', 'English'])
40 print(student.model_dump())
41 #> {'courses': 'Math Chemistry English'}
42 ```
44 Attributes:
45 func: The serializer function.
46 return_type: The return type for the function. If omitted it will be inferred from the type annotation.
47 when_used: Determines when this serializer should be used. Accepts a string with values `'always'`,
48 `'unless-none'`, `'json'`, and `'json-unless-none'`. Defaults to 'always'.
49 """
51 func: core_schema.SerializerFunction
52 return_type: TypeForm[Any] = PydanticUndefined # pyright: ignore[reportAssignmentType]
53 when_used: WhenUsed = 'always'
55 def __get_pydantic_core_schema__(self, source_type: Any, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema:
56 """Gets the Pydantic core schema.
58 Args:
59 source_type: The source type.
60 handler: The `GetCoreSchemaHandler` instance.
62 Returns:
63 The Pydantic core schema.
64 """
65 schema = handler(source_type)
66 if self.return_type is not PydanticUndefined:
67 return_type = self.return_type
68 else:
69 try:
70 # Do not pass in globals as the function could be defined in a different module.
71 # Instead, let `get_callable_return_type` infer the globals to use, but still pass
72 # in locals that may contain a parent/rebuild namespace:
73 return_type = _decorators.get_callable_return_type(
74 self.func,
75 localns=handler._get_types_namespace().locals,
76 )
77 except NameError as e:
78 raise PydanticUndefinedAnnotation.from_name_error(e) from e
80 return_schema = None if return_type is PydanticUndefined else handler.generate_schema(return_type)
81 schema['serialization'] = core_schema.plain_serializer_function_ser_schema(
82 function=self.func,
83 info_arg=_decorators.inspect_annotated_serializer(self.func, 'plain'),
84 return_schema=return_schema,
85 when_used=self.when_used,
86 )
87 return schema
90@dataclasses.dataclass(slots=True, frozen=True)
91class WrapSerializer:
92 """Wrap serializers receive the raw inputs along with a handler function that applies the standard serialization
93 logic, and can modify the resulting value before returning it as the final output of serialization.
95 For example, here's a scenario in which a wrap serializer transforms timezones to UTC **and** utilizes the existing `datetime` serialization logic.
97 ```python
98 from datetime import datetime, timezone
99 from typing import Annotated, Any
101 from pydantic import (
102 BaseModel,
103 SerializationInfo,
104 SerializerFunctionWrapHandler,
105 WrapSerializer,
106 )
108 def convert_to_utc(
109 value: Any, handler: SerializerFunctionWrapHandler, info: SerializationInfo
110 ):
111 # Note that `handler` can actually help serialize the `value` for
112 # further custom serialization in case it's a subclass.
113 serialized = handler(value)
114 if info.mode == 'json':
115 return datetime.fromisoformat(serialized).astimezone(timezone.utc)
116 else:
117 return serialized.astimezone(timezone.utc)
119 UTCDatetime = Annotated[datetime, WrapSerializer(convert_to_utc)]
121 class EventModel(BaseModel):
122 event_datetime: UTCDatetime
124 event = EventModel(event_datetime='2024-01-01T07:00:00-08:00')
125 print(event.model_dump())
126 '''
127 {
128 'event_datetime': datetime.datetime(
129 2024, 1, 1, 15, 0, tzinfo=datetime.timezone.utc
130 )
131 }
132 '''
134 print(event.model_dump_json())
135 #> {"event_datetime":"2024-01-01T15:00:00Z"}
136 ```
138 Attributes:
139 func: The serializer function to be wrapped.
140 return_type: The return type for the function. If omitted it will be inferred from the type annotation.
141 when_used: Determines when this serializer should be used. Accepts a string with values `'always'`,
142 `'unless-none'`, `'json'`, and `'json-unless-none'`. Defaults to 'always'.
143 """
145 func: core_schema.WrapSerializerFunction
146 return_type: TypeForm[Any] = PydanticUndefined # pyright: ignore[reportAssignmentType]
147 when_used: WhenUsed = 'always'
149 def __get_pydantic_core_schema__(self, source_type: Any, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema:
150 """This method is used to get the Pydantic core schema of the class.
152 Args:
153 source_type: Source type.
154 handler: Core schema handler.
156 Returns:
157 The generated core schema of the class.
158 """
159 schema = handler(source_type)
160 if self.return_type is not PydanticUndefined:
161 return_type = self.return_type
162 else:
163 try:
164 # Do not pass in globals as the function could be defined in a different module.
165 # Instead, let `get_callable_return_type` infer the globals to use, but still pass
166 # in locals that may contain a parent/rebuild namespace:
167 return_type = _decorators.get_callable_return_type(
168 self.func,
169 localns=handler._get_types_namespace().locals,
170 )
171 except NameError as e:
172 raise PydanticUndefinedAnnotation.from_name_error(e) from e
174 return_schema = None if return_type is PydanticUndefined else handler.generate_schema(return_type)
175 schema['serialization'] = core_schema.wrap_serializer_function_ser_schema(
176 function=self.func,
177 info_arg=_decorators.inspect_annotated_serializer(self.func, 'wrap'),
178 return_schema=return_schema,
179 when_used=self.when_used,
180 )
181 return schema
184if TYPE_CHECKING:
185 _Partial: TypeAlias = 'partial[Any] | partialmethod[Any]'
187 FieldPlainSerializer: TypeAlias = 'core_schema.SerializerFunction | _Partial'
188 """A field serializer method or function in `plain` mode."""
190 FieldWrapSerializer: TypeAlias = 'core_schema.WrapSerializerFunction | _Partial'
191 """A field serializer method or function in `wrap` mode."""
193 FieldSerializer: TypeAlias = 'FieldPlainSerializer | FieldWrapSerializer'
194 """A field serializer method or function."""
196 _FieldPlainSerializerT = TypeVar('_FieldPlainSerializerT', bound=FieldPlainSerializer)
197 _FieldWrapSerializerT = TypeVar('_FieldWrapSerializerT', bound=FieldWrapSerializer)
200@overload
201def field_serializer(
202 field: str,
203 /,
204 *fields: str,
205 mode: Literal['wrap'],
206 return_type: TypeForm[Any] = ...,
207 when_used: WhenUsed = ...,
208 check_fields: bool | None = ...,
209) -> Callable[[_FieldWrapSerializerT], _FieldWrapSerializerT]: ...
212@overload
213def field_serializer(
214 field: str,
215 /,
216 *fields: str,
217 mode: Literal['plain'] = ...,
218 return_type: TypeForm[Any] = ...,
219 when_used: WhenUsed = ...,
220 check_fields: bool | None = ...,
221) -> Callable[[_FieldPlainSerializerT], _FieldPlainSerializerT]: ...
224def field_serializer( # noqa: D417
225 field: str,
226 /,
227 *fields: str,
228 mode: Literal['plain', 'wrap'] = 'plain',
229 return_type: TypeForm[Any] = PydanticUndefined, # pyright: ignore[reportArgumentType]
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 pydantic import BaseModel, field_serializer
243 class StudentModel(BaseModel):
244 name: str = 'Jane'
245 courses: set[str]
247 @field_serializer('courses', when_used='json')
248 def serialize_courses_in_order(self, courses: set[str]):
249 return sorted(courses)
251 student = StudentModel(courses={'Math', 'Chemistry', 'English'})
252 print(student.model_dump_json())
253 #> {"name":"Jane","courses":["Chemistry","English","Math"]}
254 ```
256 See [the usage documentation](../concepts/serialization.md#serializers) for more information.
258 Four signatures are supported for the decorated serializer:
260 - `(self, value: Any, info: FieldSerializationInfo)`
261 - `(self, value: Any, nxt: SerializerFunctionWrapHandler, info: FieldSerializationInfo)`
262 - `(value: Any, info: SerializationInfo)`
263 - `(value: Any, nxt: SerializerFunctionWrapHandler, info: SerializationInfo)`
265 Args:
266 *fields: The field names the serializer should apply to.
267 mode: The serialization mode.
269 - `plain` means the function will be called instead of the default serialization logic,
270 - `wrap` means the function will be called with an argument to optionally call the
271 default serialization logic.
272 return_type: Optional return type for the function, if omitted it will be inferred from the type annotation.
273 when_used: Determines the serializer will be used for serialization.
274 check_fields: Whether to check that the fields actually exist on the model.
276 Raises:
277 PydanticUserError:
278 - If the decorator is used without any arguments (at least one field name must be provided).
279 - If the provided field names are not strings.
280 """
281 if callable(field) or isinstance(field, classmethod):
282 raise PydanticUserError(
283 'The `@field_serializer` decorator cannot be used without arguments, at least one field must be provided. '
284 "For example: `@field_serializer('<field_name>', ...)`.",
285 code='decorator-missing-arguments',
286 )
288 fields = field, *fields
289 if not all(isinstance(field, str) for field in fields):
290 raise PydanticUserError(
291 'The provided field names to the `@field_serializer` decorator should be strings. '
292 "For example: `@field_serializer('<field_name_1>', '<field_name_2>', ...).`",
293 code='decorator-invalid-fields',
294 )
296 def dec(f: FieldSerializer) -> _decorators.PydanticDescriptorProxy[Any]:
297 dec_info = _decorators.FieldSerializerDecoratorInfo(
298 fields=fields,
299 mode=mode,
300 return_type=return_type,
301 when_used=when_used,
302 check_fields=check_fields,
303 )
304 return _decorators.PydanticDescriptorProxy(f, dec_info) # pyright: ignore[reportArgumentType]
306 return dec # pyright: ignore[reportReturnType]
309if TYPE_CHECKING:
310 # The first argument in the following callables represent the `self` type:
312 ModelPlainSerializerWithInfo: TypeAlias = Callable[[Any, SerializationInfo[Any]], Any]
313 """A model serializer method with the `info` argument, in `plain` mode."""
315 ModelPlainSerializerWithoutInfo: TypeAlias = Callable[[Any], Any]
316 """A model serializer method without the `info` argument, in `plain` mode."""
318 ModelPlainSerializer: TypeAlias = 'ModelPlainSerializerWithInfo | ModelPlainSerializerWithoutInfo'
319 """A model serializer method in `plain` mode."""
321 ModelWrapSerializerWithInfo: TypeAlias = Callable[[Any, SerializerFunctionWrapHandler, SerializationInfo[Any]], Any]
322 """A model serializer method with the `info` argument, in `wrap` mode."""
324 ModelWrapSerializerWithoutInfo: TypeAlias = Callable[[Any, SerializerFunctionWrapHandler], Any]
325 """A model serializer method without the `info` argument, in `wrap` mode."""
327 ModelWrapSerializer: TypeAlias = 'ModelWrapSerializerWithInfo | ModelWrapSerializerWithoutInfo'
328 """A model serializer method in `wrap` mode."""
330 ModelSerializer: TypeAlias = 'ModelPlainSerializer | ModelWrapSerializer'
332 _ModelPlainSerializerT = TypeVar('_ModelPlainSerializerT', bound=ModelPlainSerializer)
333 _ModelWrapSerializerT = TypeVar('_ModelWrapSerializerT', bound=ModelWrapSerializer)
336@overload
337def model_serializer(f: _ModelPlainSerializerT, /) -> _ModelPlainSerializerT: ...
340@overload
341def model_serializer(
342 *, mode: Literal['wrap'], when_used: WhenUsed = 'always', return_type: TypeForm[Any] = ...
343) -> Callable[[_ModelWrapSerializerT], _ModelWrapSerializerT]: ...
346@overload
347def model_serializer(
348 *,
349 mode: Literal['plain'] = ...,
350 when_used: WhenUsed = 'always',
351 return_type: TypeForm[Any] = ...,
352) -> Callable[[_ModelPlainSerializerT], _ModelPlainSerializerT]: ...
355def model_serializer( # pyright: ignore[reportInconsistentOverload]
356 f: _ModelPlainSerializerT | _ModelWrapSerializerT | None = None,
357 /,
358 *,
359 mode: Literal['plain', 'wrap'] = 'plain',
360 when_used: WhenUsed = 'always',
361 return_type: TypeForm[Any] = PydanticUndefined, # pyright: ignore[reportArgumentType]
362) -> (
363 _ModelPlainSerializerT
364 | Callable[[_ModelWrapSerializerT], _ModelWrapSerializerT]
365 | Callable[[_ModelPlainSerializerT], _ModelPlainSerializerT]
366):
367 """Decorator that enables custom model serialization.
369 This is useful when a model need to be serialized in a customized manner, allowing for flexibility beyond just specific fields.
371 An example would be to serialize temperature to the same temperature scale, such as degrees Celsius.
373 ```python
374 from typing import Literal
376 from pydantic import BaseModel, model_serializer
378 class TemperatureModel(BaseModel):
379 unit: Literal['C', 'F']
380 value: int
382 @model_serializer()
383 def serialize_model(self):
384 if self.unit == 'F':
385 return {'unit': 'C', 'value': int((self.value - 32) / 1.8)}
386 return {'unit': self.unit, 'value': self.value}
388 temperature = TemperatureModel(unit='F', value=212)
389 print(temperature.model_dump())
390 #> {'unit': 'C', 'value': 100}
391 ```
393 Two signatures are supported for `mode='plain'`, which is the default:
395 - `(self)`
396 - `(self, info: SerializationInfo)`
398 And two other signatures for `mode='wrap'`:
400 - `(self, nxt: SerializerFunctionWrapHandler)`
401 - `(self, nxt: SerializerFunctionWrapHandler, info: SerializationInfo)`
403 See [the usage documentation](../concepts/serialization.md#serializers) for more information.
405 Args:
406 f: The function to be decorated.
407 mode: The serialization mode.
409 - `'plain'` means the function will be called instead of the default serialization logic
410 - `'wrap'` means the function will be called with an argument to optionally call the default
411 serialization logic.
412 when_used: Determines when this serializer should be used.
413 return_type: The return type for the function. If omitted it will be inferred from the type annotation.
415 Returns:
416 The decorator function.
417 """
419 def dec(f: ModelSerializer) -> _decorators.PydanticDescriptorProxy[Any]:
420 dec_info = _decorators.ModelSerializerDecoratorInfo(mode=mode, return_type=return_type, when_used=when_used)
421 return _decorators.PydanticDescriptorProxy(f, dec_info)
423 if f is None:
424 return dec # pyright: ignore[reportReturnType]
425 else:
426 return dec(f) # pyright: ignore[reportReturnType]
429AnyType = TypeVar('AnyType')
432if TYPE_CHECKING:
433 SerializeAsAny = Annotated[AnyType, ...] # SerializeAsAny[list[str]] will be treated by type checkers as list[str]
434 """Annotation used to mark a type as having duck-typing serialization behavior.
436 See [usage documentation](../concepts/serialization.md#serializing-with-duck-typing) for more details.
437 """
438else:
440 @dataclasses.dataclass(slots=True)
441 class SerializeAsAny:
442 """Annotation used to mark a type as having duck-typing serialization behavior.
444 See [usage documentation](../concepts/serialization.md#serializing-with-duck-typing) for more details.
445 """
447 def __class_getitem__(cls, item: Any) -> Any:
448 return Annotated[item, SerializeAsAny()]
450 def __get_pydantic_core_schema__(
451 self, source_type: Any, handler: GetCoreSchemaHandler
452 ) -> core_schema.CoreSchema:
453 schema = handler(source_type)
454 schema_to_update = schema
455 while schema_to_update['type'] == 'definitions':
456 schema_to_update = schema_to_update.copy()
457 schema_to_update = schema_to_update['schema']
458 schema_to_update['serialization'] = core_schema.simple_ser_schema('any')
459 return schema
461 __hash__ = object.__hash__