Coverage for pydantic/functional_serializers.py: 96.10%
65 statements
« prev ^ index » next coverage.py v7.5.3, created at 2024-06-21 17:00 +0000
1"""This module contains related classes and functions for serialization."""
3from __future__ import annotations 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
5import dataclasses 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
6from functools import partialmethod 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
7from typing import TYPE_CHECKING, Any, Callable, TypeVar, Union, overload 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
9from pydantic_core import PydanticUndefined, core_schema 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
10from pydantic_core import core_schema as _core_schema 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
11from typing_extensions import Annotated, Literal, TypeAlias 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
13from . import PydanticUndefinedAnnotation 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
14from ._internal import _decorators, _internal_dataclass 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
15from .annotated_handlers import GetCoreSchemaHandler 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
18@dataclasses.dataclass(**_internal_dataclass.slots_true, frozen=True) 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
19class PlainSerializer: 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
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 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
52 return_type: Any = PydanticUndefined 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
53 when_used: Literal['always', 'unless-none', 'json', 'json-unless-none'] = 'always' 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
55 def __get_pydantic_core_schema__(self, source_type: Any, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema: 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
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) 1ijklmnopabghqrstuvwxcdyzABCDEFef
66 try: 1ijklmnopabghqrstuvwxcdyzABCDEFef
67 return_type = _decorators.get_function_return_type( 1ijklmnopabghqrstuvwxcdyzABCDEFef
68 self.func, self.return_type, handler._get_types_namespace()
69 )
70 except NameError as e: 1ijklmnopabghqrstuvwxcdyzABCDEFef
71 raise PydanticUndefinedAnnotation.from_name_error(e) from e 1ijklmnopabghqrstuvwxcdyzABCDEFef
72 return_schema = None if return_type is PydanticUndefined else handler.generate_schema(return_type) 1ijklmnopabghqrstuvwxcdyzABCDEFef
73 schema['serialization'] = core_schema.plain_serializer_function_ser_schema( 1ijklmnopabghqrstuvwxcdyzABCDEFef
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 1ijklmnopabghqrstuvwxcdyzABCDEFef
82@dataclasses.dataclass(**_internal_dataclass.slots_true, frozen=True) 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
83class WrapSerializer: 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
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 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
148 return_type: Any = PydanticUndefined 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
149 when_used: Literal['always', 'unless-none', 'json', 'json-unless-none'] = 'always' 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
151 def __get_pydantic_core_schema__(self, source_type: Any, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema: 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
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) 1ijklmnopabghqrstuvwxcdyzABCDEFef
162 try: 1ijklmnopabghqrstuvwxcdyzABCDEFef
163 return_type = _decorators.get_function_return_type( 1ijklmnopabghqrstuvwxcdyzABCDEFef
164 self.func, self.return_type, handler._get_types_namespace()
165 )
166 except NameError as e: 1ijklmnopabghqrstuvwxcdyzABCDEFef
167 raise PydanticUndefinedAnnotation.from_name_error(e) from e 1ijklmnopabghqrstuvwxcdyzABCDEFef
168 return_schema = None if return_type is PydanticUndefined else handler.generate_schema(return_type) 1ijklmnopabghqrstuvwxcdyzABCDEFef
169 schema['serialization'] = core_schema.wrap_serializer_function_ser_schema( 1ijklmnopabghqrstuvwxcdyzABCDEFef
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 1ijklmnopabghqrstuvwxcdyzABCDEFef
178if TYPE_CHECKING: 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
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 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
187def field_serializer( 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
188 field: str, 1abghcdGef
189 /,
190 *fields: str, 1abghcdGef
191 return_type: Any = ..., 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
192 when_used: Literal['always', 'unless-none', 'json', 'json-unless-none'] = ..., 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
193 check_fields: bool | None = ..., 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
194) -> Callable[[_PlainSerializeMethodType], _PlainSerializeMethodType]: ... 1abghcdGef
197@overload 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
198def field_serializer( 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
199 field: str, 1abghcdGef
200 /,
201 *fields: str, 1abghcdGef
202 mode: Literal['plain'], 1abghcdGef
203 return_type: Any = ..., 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
204 when_used: Literal['always', 'unless-none', 'json', 'json-unless-none'] = ..., 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
205 check_fields: bool | None = ..., 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
206) -> Callable[[_PlainSerializeMethodType], _PlainSerializeMethodType]: ... 1abghcdGef
209@overload 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
210def field_serializer( 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
211 field: str, 1abghcdGef
212 /,
213 *fields: str, 1abghcdGef
214 mode: Literal['wrap'], 1abghcdGef
215 return_type: Any = ..., 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
216 when_used: Literal['always', 'unless-none', 'json', 'json-unless-none'] = ..., 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
217 check_fields: bool | None = ..., 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
218) -> Callable[[_WrapSerializeMethodType], _WrapSerializeMethodType]: ... 1abghcdGef
221def field_serializer( 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
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( 1ijklmnopabghqrstuvwxcdyzABCDEFef
275 f: Callable[..., Any] | staticmethod[Any, Any] | classmethod[Any, Any, Any],
276 ) -> _decorators.PydanticDescriptorProxy[Any]:
277 dec_info = _decorators.FieldSerializerDecoratorInfo( 1ijklmnopabghqrstuvwxcdyzABCDEFef
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) 1ijklmnopabghqrstuvwxcdyzABCDEFef
286 return dec 1ijklmnopabghqrstuvwxcdyzABCDEFef
289FuncType = TypeVar('FuncType', bound=Callable[..., Any]) 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
292@overload 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
293def model_serializer(__f: FuncType) -> FuncType: ... 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
296@overload 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
297def model_serializer( 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
298 *,
299 mode: Literal['plain', 'wrap'] = ..., 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
300 when_used: Literal['always', 'unless-none', 'json', 'json-unless-none'] = 'always', 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
301 return_type: Any = ..., 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
302) -> Callable[[FuncType], FuncType]: ... 1abghcdGef
305def model_serializer( 1ijklmnopabqrstuvwxcdHIJKLMNOGyzABCDEFef
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]: 1ijklmnopabghqrstuvwxcdyzABCDEFef
356 dec_info = _decorators.ModelSerializerDecoratorInfo(mode=mode, return_type=return_type, when_used=when_used) 1ijklmnopabghqrstuvwxcdyzABCDEFef
357 return _decorators.PydanticDescriptorProxy(f, dec_info) 1ijklmnopabghqrstuvwxcdyzABCDEFef
359 if f is None: 1ijklmnopabghqrstuvwxcdyzABCDEFef
360 return dec 1ijklmnopabghqrstuvwxcdyzABCDEFef
361 else:
362 return dec(f) # type: ignore 1ijklmnopabghqrstuvwxcdyzABCDEFef
365AnyType = TypeVar('AnyType') 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
368if TYPE_CHECKING: 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
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) 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
378 class SerializeAsAny: # noqa: D101 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
379 def __class_getitem__(cls, item: Any) -> Any: 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
380 return Annotated[item, SerializeAsAny()] 1ijklmnopabghqrstuvwxcdyzABCDEFef
382 def __get_pydantic_core_schema__( 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef
383 self, source_type: Any, handler: GetCoreSchemaHandler
384 ) -> core_schema.CoreSchema:
385 schema = handler(source_type) 1ijklmnopabghqrstuvwxcdyzABCDEFef
386 schema_to_update = schema 1ijklmnopabghqrstuvwxcdyzABCDEFef
387 while schema_to_update['type'] == 'definitions': 387 ↛ 388line 387 didn't jump to line 388, because the condition on line 387 was never true1ijklmnopabghqrstuvwxcdyzABCDEFef
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( 1ijklmnopabghqrstuvwxcdyzABCDEFef
391 lambda x, h: h(x), schema=core_schema.any_schema()
392 )
393 return schema 1ijklmnopabghqrstuvwxcdyzABCDEFef
395 __hash__ = object.__hash__ 1ijklmnopabghqrstuvwxcdHIJKLMNOGyzABCDEFef