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

1"""This module contains related classes and functions for serialization.""" 

2 

3from __future__ import annotations 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl

4 

5import dataclasses 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl

6from functools import partialmethod 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl

7from typing import TYPE_CHECKING, Any, Callable, TypeVar, Union, overload 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl

8 

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

12 

13from . import PydanticUndefinedAnnotation 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl

14from ._internal import _decorators, _internal_dataclass 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl

15from .annotated_handlers import GetCoreSchemaHandler 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl

16 

17 

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. 

21 

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. 

24 

25 ```python 

26 from typing import List 

27 

28 from typing_extensions import Annotated 

29 

30 from pydantic import BaseModel, PlainSerializer 

31 

32 CustomStr = Annotated[ 

33 List, PlainSerializer(lambda x: ' '.join(x), return_type=str) 

34 ] 

35 

36 class StudentModel(BaseModel): 

37 courses: CustomStr 

38 

39 student = StudentModel(courses=['Math', 'Chemistry', 'English']) 

40 print(student.model_dump()) 

41 #> {'courses': 'Math Chemistry English'} 

42 ``` 

43 

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 """ 

50 

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

54 

55 def __get_pydantic_core_schema__(self, source_type: Any, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema: 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl

56 """Gets the Pydantic core schema. 

57 

58 Args: 

59 source_type: The source type. 

60 handler: The `GetCoreSchemaHandler` instance. 

61 

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

80 

81 

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. 

86 

87 For example, here's a scenario in which a wrap serializer transforms timezones to UTC **and** utilizes the existing `datetime` serialization logic. 

88 

89 ```python 

90 from datetime import datetime, timezone 

91 from typing import Any, Dict 

92 

93 from typing_extensions import Annotated 

94 

95 from pydantic import BaseModel, WrapSerializer 

96 

97 class EventDatetime(BaseModel): 

98 start: datetime 

99 end: datetime 

100 

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()} 

110 

111 UTCEventDatetime = Annotated[EventDatetime, WrapSerializer(convert_to_utc)] 

112 

113 class EventModel(BaseModel): 

114 event_datetime: UTCEventDatetime 

115 

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 ''' 

133 

134 print(event.model_dump_json()) 

135 ''' 

136 {"event_datetime":{"start":"2024-01-01T15:00:00Z","end":"2024-01-03T14:00:00Z"}} 

137 ''' 

138 ``` 

139 

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 """ 

146 

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

150 

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. 

153 

154 Args: 

155 source_type: Source type. 

156 handler: Core schema handler. 

157 

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

176 

177 

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) 

184 

185 

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

195 

196 

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

207 

208 

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

219 

220 

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. 

229 

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. 

231 

232 ```python 

233 from typing import Set 

234 

235 from pydantic import BaseModel, field_serializer 

236 

237 class StudentModel(BaseModel): 

238 name: str = 'Jane' 

239 courses: Set[str] 

240 

241 @field_serializer('courses', when_used='json') 

242 def serialize_courses_in_order(courses: Set[str]): 

243 return sorted(courses) 

244 

245 student = StudentModel(courses={'Math', 'Chemistry', 'English'}) 

246 print(student.model_dump_json()) 

247 #> {"name":"Jane","courses":["Chemistry","English","Math"]} 

248 ``` 

249 

250 See [Custom serializers](../concepts/serialization.md#custom-serializers) for more information. 

251 

252 Four signatures are supported: 

253 

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)` 

258 

259 Args: 

260 fields: Which field(s) the method should be called on. 

261 mode: The serialization mode. 

262 

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. 

269 

270 Returns: 

271 The decorator function. 

272 """ 

273 

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

285 

286 return dec 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl

287 

288 

289FuncType = TypeVar('FuncType', bound=Callable[..., Any]) 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl

290 

291 

292@overload 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl

293def model_serializer(__f: FuncType) -> FuncType: ... 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl

294 

295 

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

303 

304 

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. 

314 

315 This is useful when a model need to be serialized in a customized manner, allowing for flexibility beyond just specific fields. 

316 

317 An example would be to serialize temperature to the same temperature scale, such as degrees Celsius. 

318 

319 ```python 

320 from typing import Literal 

321 

322 from pydantic import BaseModel, model_serializer 

323 

324 class TemperatureModel(BaseModel): 

325 unit: Literal['C', 'F'] 

326 value: int 

327 

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} 

333 

334 temperature = TemperatureModel(unit='F', value=212) 

335 print(temperature.model_dump()) 

336 #> {'unit': 'C', 'value': 100} 

337 ``` 

338 

339 See [Custom serializers](../concepts/serialization.md#custom-serializers) for more information. 

340 

341 Args: 

342 f: The function to be decorated. 

343 mode: The serialization mode. 

344 

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. 

350 

351 Returns: 

352 The decorator function. 

353 """ 

354 

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

358 

359 if f is None: 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl

360 return dec 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl

361 else: 

362 return dec(f) # type: ignore 1opqrstuvabcdmnwxyzABCDefghEFGHIJKLijkl

363 

364 

365AnyType = TypeVar('AnyType') 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl

366 

367 

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: 

376 

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

381 

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

394 

395 __hash__ = object.__hash__ 1opqrstuvabcdmnwxyzABCDefghOPQRSTUVMNEFGHIJKLijkl