Coverage for pydantic/functional_serializers.py: 96.10%
65 statements
« prev ^ index » next coverage.py v7.5.4, created at 2024-07-03 19:29 +0000
« prev ^ index » next coverage.py v7.5.4, created at 2024-07-03 19:29 +0000
1"""This module contains related classes and functions for serialization."""
3from __future__ import annotations 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
5import dataclasses 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
6from functools import partialmethod 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
7from typing import TYPE_CHECKING, Any, Callable, TypeVar, Union, overload 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
9from pydantic_core import PydanticUndefined, core_schema 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
10from pydantic_core import core_schema as _core_schema 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
11from typing_extensions import Annotated, Literal, TypeAlias 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
13from . import PydanticUndefinedAnnotation 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
14from ._internal import _decorators, _internal_dataclass 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
15from .annotated_handlers import GetCoreSchemaHandler 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
18@dataclasses.dataclass(**_internal_dataclass.slots_true, frozen=True) 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
19class PlainSerializer: 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
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 List
28 from typing_extensions 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 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
52 return_type: Any = PydanticUndefined 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
53 when_used: Literal['always', 'unless-none', 'json', 'json-unless-none'] = 'always' 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
55 def __get_pydantic_core_schema__(self, source_type: Any, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema: 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
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) 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
66 try: 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
67 return_type = _decorators.get_function_return_type( 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
68 self.func, self.return_type, handler._get_types_namespace()
69 )
70 except NameError as e: 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
71 raise PydanticUndefinedAnnotation.from_name_error(e) from e 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
72 return_schema = None if return_type is PydanticUndefined else handler.generate_schema(return_type) 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
73 schema['serialization'] = core_schema.plain_serializer_function_ser_schema( 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
74 function=self.func,
75 info_arg=_decorators.inspect_annotated_serializer(self.func, 'plain'),
76 return_schema=return_schema,
77 when_used=self.when_used,
78 )
79 return schema 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
82@dataclasses.dataclass(**_internal_dataclass.slots_true, frozen=True) 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
83class WrapSerializer: 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
84 """Wrap serializers receive the raw inputs along with a handler function that applies the standard serialization
85 logic, and can modify the resulting value before returning it as the final output of serialization.
87 For example, here's a scenario in which a wrap serializer transforms timezones to UTC **and** utilizes the existing `datetime` serialization logic.
89 ```python
90 from datetime import datetime, timezone
91 from typing import Any, Dict
93 from typing_extensions import Annotated
95 from pydantic import BaseModel, WrapSerializer
97 class EventDatetime(BaseModel):
98 start: datetime
99 end: datetime
101 def convert_to_utc(value: Any, handler, info) -> Dict[str, datetime]:
102 # Note that `helper` can actually help serialize the `value` for further custom serialization in case it's a subclass.
103 partial_result = handler(value, info)
104 if info.mode == 'json':
105 return {
106 k: datetime.fromisoformat(v).astimezone(timezone.utc)
107 for k, v in partial_result.items()
108 }
109 return {k: v.astimezone(timezone.utc) for k, v in partial_result.items()}
111 UTCEventDatetime = Annotated[EventDatetime, WrapSerializer(convert_to_utc)]
113 class EventModel(BaseModel):
114 event_datetime: UTCEventDatetime
116 dt = EventDatetime(
117 start='2024-01-01T07:00:00-08:00', end='2024-01-03T20:00:00+06:00'
118 )
119 event = EventModel(event_datetime=dt)
120 print(event.model_dump())
121 '''
122 {
123 'event_datetime': {
124 'start': datetime.datetime(
125 2024, 1, 1, 15, 0, tzinfo=datetime.timezone.utc
126 ),
127 'end': datetime.datetime(
128 2024, 1, 3, 14, 0, tzinfo=datetime.timezone.utc
129 ),
130 }
131 }
132 '''
134 print(event.model_dump_json())
135 '''
136 {"event_datetime":{"start":"2024-01-01T15:00:00Z","end":"2024-01-03T14:00:00Z"}}
137 '''
138 ```
140 Attributes:
141 func: The serializer function to be wrapped.
142 return_type: The return type for the function. If omitted it will be inferred from the type annotation.
143 when_used: Determines when this serializer should be used. Accepts a string with values `'always'`,
144 `'unless-none'`, `'json'`, and `'json-unless-none'`. Defaults to 'always'.
145 """
147 func: core_schema.WrapSerializerFunction 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
148 return_type: Any = PydanticUndefined 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
149 when_used: Literal['always', 'unless-none', 'json', 'json-unless-none'] = 'always' 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
151 def __get_pydantic_core_schema__(self, source_type: Any, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema: 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
152 """This method is used to get the Pydantic core schema of the class.
154 Args:
155 source_type: Source type.
156 handler: Core schema handler.
158 Returns:
159 The generated core schema of the class.
160 """
161 schema = handler(source_type) 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
162 try: 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
163 return_type = _decorators.get_function_return_type( 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
164 self.func, self.return_type, handler._get_types_namespace()
165 )
166 except NameError as e: 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
167 raise PydanticUndefinedAnnotation.from_name_error(e) from e 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
168 return_schema = None if return_type is PydanticUndefined else handler.generate_schema(return_type) 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
169 schema['serialization'] = core_schema.wrap_serializer_function_ser_schema( 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
170 function=self.func,
171 info_arg=_decorators.inspect_annotated_serializer(self.func, 'wrap'),
172 return_schema=return_schema,
173 when_used=self.when_used,
174 )
175 return schema 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
178if TYPE_CHECKING: 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
179 _PartialClsOrStaticMethod: TypeAlias = Union[classmethod[Any, Any, Any], staticmethod[Any, Any], partialmethod[Any]]
180 _PlainSerializationFunction = Union[_core_schema.SerializerFunction, _PartialClsOrStaticMethod]
181 _WrapSerializationFunction = Union[_core_schema.WrapSerializerFunction, _PartialClsOrStaticMethod]
182 _PlainSerializeMethodType = TypeVar('_PlainSerializeMethodType', bound=_PlainSerializationFunction)
183 _WrapSerializeMethodType = TypeVar('_WrapSerializeMethodType', bound=_WrapSerializationFunction)
186@overload 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
187def field_serializer( 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
188 field: str, 1abcdmnefghMNijkl
189 /,
190 *fields: str, 1abcdmnefghMNijkl
191 return_type: Any = ..., 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
192 when_used: Literal['always', 'unless-none', 'json', 'json-unless-none'] = ..., 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
193 check_fields: bool | None = ..., 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
194) -> Callable[[_PlainSerializeMethodType], _PlainSerializeMethodType]: ... 1abcdmnefghMNijkl
197@overload 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
198def field_serializer( 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
199 field: str, 1abcdmnefghMNijkl
200 /,
201 *fields: str, 1abcdmnefghMNijkl
202 mode: Literal['plain'], 1abcdmnefghMNijkl
203 return_type: Any = ..., 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
204 when_used: Literal['always', 'unless-none', 'json', 'json-unless-none'] = ..., 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
205 check_fields: bool | None = ..., 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
206) -> Callable[[_PlainSerializeMethodType], _PlainSerializeMethodType]: ... 1abcdmnefghMNijkl
209@overload 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
210def field_serializer( 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
211 field: str, 1abcdmnefghMNijkl
212 /,
213 *fields: str, 1abcdmnefghMNijkl
214 mode: Literal['wrap'], 1abcdmnefghMNijkl
215 return_type: Any = ..., 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
216 when_used: Literal['always', 'unless-none', 'json', 'json-unless-none'] = ..., 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
217 check_fields: bool | None = ..., 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
218) -> Callable[[_WrapSerializeMethodType], _WrapSerializeMethodType]: ... 1abcdmnefghMNijkl
221def field_serializer( 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
222 *fields: str,
223 mode: Literal['plain', 'wrap'] = 'plain',
224 return_type: Any = PydanticUndefined,
225 when_used: Literal['always', 'unless-none', 'json', 'json-unless-none'] = 'always',
226 check_fields: bool | None = None,
227) -> Callable[[Any], Any]:
228 """Decorator that enables custom field serialization.
230 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.
232 ```python
233 from typing import Set
235 from pydantic import BaseModel, field_serializer
237 class StudentModel(BaseModel):
238 name: str = 'Jane'
239 courses: Set[str]
241 @field_serializer('courses', when_used='json')
242 def serialize_courses_in_order(courses: Set[str]):
243 return sorted(courses)
245 student = StudentModel(courses={'Math', 'Chemistry', 'English'})
246 print(student.model_dump_json())
247 #> {"name":"Jane","courses":["Chemistry","English","Math"]}
248 ```
250 See [Custom serializers](../concepts/serialization.md#custom-serializers) for more information.
252 Four signatures are supported:
254 - `(self, value: Any, info: FieldSerializationInfo)`
255 - `(self, value: Any, nxt: SerializerFunctionWrapHandler, info: FieldSerializationInfo)`
256 - `(value: Any, info: SerializationInfo)`
257 - `(value: Any, nxt: SerializerFunctionWrapHandler, info: SerializationInfo)`
259 Args:
260 fields: Which field(s) the method should be called on.
261 mode: The serialization mode.
263 - `plain` means the function will be called instead of the default serialization logic,
264 - `wrap` means the function will be called with an argument to optionally call the
265 default serialization logic.
266 return_type: Optional return type for the function, if omitted it will be inferred from the type annotation.
267 when_used: Determines the serializer will be used for serialization.
268 check_fields: Whether to check that the fields actually exist on the model.
270 Returns:
271 The decorator function.
272 """
274 def dec( 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
275 f: Callable[..., Any] | staticmethod[Any, Any] | classmethod[Any, Any, Any],
276 ) -> _decorators.PydanticDescriptorProxy[Any]:
277 dec_info = _decorators.FieldSerializerDecoratorInfo( 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
278 fields=fields,
279 mode=mode,
280 return_type=return_type,
281 when_used=when_used,
282 check_fields=check_fields,
283 )
284 return _decorators.PydanticDescriptorProxy(f, dec_info) 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
286 return dec 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
289FuncType = TypeVar('FuncType', bound=Callable[..., Any]) 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
292@overload 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
293def model_serializer(__f: FuncType) -> FuncType: ... 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
296@overload 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
297def model_serializer( 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
298 *,
299 mode: Literal['plain', 'wrap'] = ..., 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
300 when_used: Literal['always', 'unless-none', 'json', 'json-unless-none'] = 'always', 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
301 return_type: Any = ..., 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
302) -> Callable[[FuncType], FuncType]: ... 1abcdmnefghMNijkl
305def model_serializer( 1opqrstuvabcdwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
306 f: Callable[..., Any] | None = None,
307 /,
308 *,
309 mode: Literal['plain', 'wrap'] = 'plain',
310 when_used: Literal['always', 'unless-none', 'json', 'json-unless-none'] = 'always',
311 return_type: Any = PydanticUndefined,
312) -> Callable[[Any], Any]:
313 """Decorator that enables custom model serialization.
315 This is useful when a model need to be serialized in a customized manner, allowing for flexibility beyond just specific fields.
317 An example would be to serialize temperature to the same temperature scale, such as degrees Celsius.
319 ```python
320 from typing import Literal
322 from pydantic import BaseModel, model_serializer
324 class TemperatureModel(BaseModel):
325 unit: Literal['C', 'F']
326 value: int
328 @model_serializer()
329 def serialize_model(self):
330 if self.unit == 'F':
331 return {'unit': 'C', 'value': int((self.value - 32) / 1.8)}
332 return {'unit': self.unit, 'value': self.value}
334 temperature = TemperatureModel(unit='F', value=212)
335 print(temperature.model_dump())
336 #> {'unit': 'C', 'value': 100}
337 ```
339 See [Custom serializers](../concepts/serialization.md#custom-serializers) for more information.
341 Args:
342 f: The function to be decorated.
343 mode: The serialization mode.
345 - `'plain'` means the function will be called instead of the default serialization logic
346 - `'wrap'` means the function will be called with an argument to optionally call the default
347 serialization logic.
348 when_used: Determines when this serializer should be used.
349 return_type: The return type for the function. If omitted it will be inferred from the type annotation.
351 Returns:
352 The decorator function.
353 """
355 def dec(f: Callable[..., Any]) -> _decorators.PydanticDescriptorProxy[Any]: 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
356 dec_info = _decorators.ModelSerializerDecoratorInfo(mode=mode, return_type=return_type, when_used=when_used) 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
357 return _decorators.PydanticDescriptorProxy(f, dec_info) 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
359 if f is None: 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
360 return dec 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
361 else:
362 return dec(f) # type: ignore 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
365AnyType = TypeVar('AnyType') 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
368if TYPE_CHECKING: 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
369 SerializeAsAny = Annotated[AnyType, ...] # SerializeAsAny[list[str]] will be treated by type checkers as list[str]
370 """Force serialization to ignore whatever is defined in the schema and instead ask the object
371 itself how it should be serialized.
372 In particular, this means that when model subclasses are serialized, fields present in the subclass
373 but not in the original schema will be included.
374 """
375else:
377 @dataclasses.dataclass(**_internal_dataclass.slots_true) 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
378 class SerializeAsAny: # noqa: D101 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
379 def __class_getitem__(cls, item: Any) -> Any: 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
380 return Annotated[item, SerializeAsAny()] 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
382 def __get_pydantic_core_schema__( 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl
383 self, source_type: Any, handler: GetCoreSchemaHandler
384 ) -> core_schema.CoreSchema:
385 schema = handler(source_type) 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
386 schema_to_update = schema 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
387 while schema_to_update['type'] == 'definitions': 387 ↛ 388line 387 didn't jump to line 388 because the condition on line 387 was never true1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
388 schema_to_update = schema_to_update.copy()
389 schema_to_update = schema_to_update['schema']
390 schema_to_update['serialization'] = core_schema.wrap_serializer_function_ser_schema( 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
391 lambda x, h: h(x), schema=core_schema.any_schema()
392 )
393 return schema 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl
395 __hash__ = object.__hash__ 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl