Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/pydantic/functional_serializers.py: 46%

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

2 

3from __future__ import annotations 

4 

5import dataclasses 

6from functools import partial, partialmethod 

7from typing import TYPE_CHECKING, Annotated, Any, Callable, Literal, TypeVar, overload 

8 

9from pydantic_core import PydanticUndefined, core_schema 

10from pydantic_core.core_schema import SerializationInfo, SerializerFunctionWrapHandler, WhenUsed 

11from typing_extensions import TypeAlias 

12 

13from . import PydanticUndefinedAnnotation 

14from ._internal import _decorators, _internal_dataclass 

15from .annotated_handlers import GetCoreSchemaHandler 

16 

17 

18@dataclasses.dataclass(**_internal_dataclass.slots_true, frozen=True) 

19class PlainSerializer: 

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 Annotated 

27 

28 from pydantic import BaseModel, PlainSerializer 

29 

30 CustomStr = Annotated[ 

31 list, PlainSerializer(lambda x: ' '.join(x), return_type=str) 

32 ] 

33 

34 class StudentModel(BaseModel): 

35 courses: CustomStr 

36 

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

38 print(student.model_dump()) 

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

40 ``` 

41 

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

48 

49 func: core_schema.SerializerFunction 

50 return_type: Any = PydanticUndefined 

51 when_used: WhenUsed = 'always' 

52 

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

54 """Gets the Pydantic core schema. 

55 

56 Args: 

57 source_type: The source type. 

58 handler: The `GetCoreSchemaHandler` instance. 

59 

60 Returns: 

61 The Pydantic core schema. 

62 """ 

63 schema = handler(source_type) 

64 if self.return_type is not PydanticUndefined: 

65 return_type = self.return_type 

66 else: 

67 try: 

68 # Do not pass in globals as the function could be defined in a different module. 

69 # Instead, let `get_callable_return_type` infer the globals to use, but still pass 

70 # in locals that may contain a parent/rebuild namespace: 

71 return_type = _decorators.get_callable_return_type( 

72 self.func, 

73 localns=handler._get_types_namespace().locals, 

74 ) 

75 except NameError as e: 

76 raise PydanticUndefinedAnnotation.from_name_error(e) from e 

77 

78 return_schema = None if return_type is PydanticUndefined else handler.generate_schema(return_type) 

79 schema['serialization'] = core_schema.plain_serializer_function_ser_schema( 

80 function=self.func, 

81 info_arg=_decorators.inspect_annotated_serializer(self.func, 'plain'), 

82 return_schema=return_schema, 

83 when_used=self.when_used, 

84 ) 

85 return schema 

86 

87 

88@dataclasses.dataclass(**_internal_dataclass.slots_true, frozen=True) 

89class WrapSerializer: 

90 """Wrap serializers receive the raw inputs along with a handler function that applies the standard serialization 

91 logic, and can modify the resulting value before returning it as the final output of serialization. 

92 

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

94 

95 ```python 

96 from datetime import datetime, timezone 

97 from typing import Annotated, Any 

98 

99 from pydantic import BaseModel, WrapSerializer 

100 

101 class EventDatetime(BaseModel): 

102 start: datetime 

103 end: datetime 

104 

105 def convert_to_utc(value: Any, handler, info) -> dict[str, datetime]: 

106 # Note that `handler` can actually help serialize the `value` for 

107 # further custom serialization in case it's a subclass. 

108 partial_result = handler(value, info) 

109 if info.mode == 'json': 

110 return { 

111 k: datetime.fromisoformat(v).astimezone(timezone.utc) 

112 for k, v in partial_result.items() 

113 } 

114 return {k: v.astimezone(timezone.utc) for k, v in partial_result.items()} 

115 

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

117 

118 class EventModel(BaseModel): 

119 event_datetime: UTCEventDatetime 

120 

121 dt = EventDatetime( 

122 start='2024-01-01T07:00:00-08:00', end='2024-01-03T20:00:00+06:00' 

123 ) 

124 event = EventModel(event_datetime=dt) 

125 print(event.model_dump()) 

126 ''' 

127 { 

128 'event_datetime': { 

129 'start': datetime.datetime( 

130 2024, 1, 1, 15, 0, tzinfo=datetime.timezone.utc 

131 ), 

132 'end': datetime.datetime( 

133 2024, 1, 3, 14, 0, tzinfo=datetime.timezone.utc 

134 ), 

135 } 

136 } 

137 ''' 

138 

139 print(event.model_dump_json()) 

140 ''' 

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

142 ''' 

143 ``` 

144 

145 Attributes: 

146 func: The serializer function to be wrapped. 

147 return_type: The return type for the function. If omitted it will be inferred from the type annotation. 

148 when_used: Determines when this serializer should be used. Accepts a string with values `'always'`, 

149 `'unless-none'`, `'json'`, and `'json-unless-none'`. Defaults to 'always'. 

150 """ 

151 

152 func: core_schema.WrapSerializerFunction 

153 return_type: Any = PydanticUndefined 

154 when_used: WhenUsed = 'always' 

155 

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

157 """This method is used to get the Pydantic core schema of the class. 

158 

159 Args: 

160 source_type: Source type. 

161 handler: Core schema handler. 

162 

163 Returns: 

164 The generated core schema of the class. 

165 """ 

166 schema = handler(source_type) 

167 if self.return_type is not PydanticUndefined: 

168 return_type = self.return_type 

169 else: 

170 try: 

171 # Do not pass in globals as the function could be defined in a different module. 

172 # Instead, let `get_callable_return_type` infer the globals to use, but still pass 

173 # in locals that may contain a parent/rebuild namespace: 

174 return_type = _decorators.get_callable_return_type( 

175 self.func, 

176 localns=handler._get_types_namespace().locals, 

177 ) 

178 except NameError as e: 

179 raise PydanticUndefinedAnnotation.from_name_error(e) from e 

180 

181 return_schema = None if return_type is PydanticUndefined else handler.generate_schema(return_type) 

182 schema['serialization'] = core_schema.wrap_serializer_function_ser_schema( 

183 function=self.func, 

184 info_arg=_decorators.inspect_annotated_serializer(self.func, 'wrap'), 

185 return_schema=return_schema, 

186 when_used=self.when_used, 

187 ) 

188 return schema 

189 

190 

191if TYPE_CHECKING: 

192 _Partial: TypeAlias = 'partial[Any] | partialmethod[Any]' 

193 

194 FieldPlainSerializer: TypeAlias = 'core_schema.SerializerFunction | _Partial' 

195 """A field serializer method or function in `plain` mode.""" 

196 

197 FieldWrapSerializer: TypeAlias = 'core_schema.WrapSerializerFunction | _Partial' 

198 """A field serializer method or function in `wrap` mode.""" 

199 

200 FieldSerializer: TypeAlias = 'FieldPlainSerializer | FieldWrapSerializer' 

201 """A field serializer method or function.""" 

202 

203 _FieldPlainSerializerT = TypeVar('_FieldPlainSerializerT', bound=FieldPlainSerializer) 

204 _FieldWrapSerializerT = TypeVar('_FieldWrapSerializerT', bound=FieldWrapSerializer) 

205 

206 

207@overload 

208def field_serializer( 

209 field: str, 

210 /, 

211 *fields: str, 

212 mode: Literal['wrap'], 

213 return_type: Any = ..., 

214 when_used: WhenUsed = ..., 

215 check_fields: bool | None = ..., 

216) -> Callable[[_FieldWrapSerializerT], _FieldWrapSerializerT]: ... 

217 

218 

219@overload 

220def field_serializer( 

221 field: str, 

222 /, 

223 *fields: str, 

224 mode: Literal['plain'] = ..., 

225 return_type: Any = ..., 

226 when_used: WhenUsed = ..., 

227 check_fields: bool | None = ..., 

228) -> Callable[[_FieldPlainSerializerT], _FieldPlainSerializerT]: ... 

229 

230 

231def field_serializer( 

232 *fields: str, 

233 mode: Literal['plain', 'wrap'] = 'plain', 

234 return_type: Any = PydanticUndefined, 

235 when_used: WhenUsed = 'always', 

236 check_fields: bool | None = None, 

237) -> ( 

238 Callable[[_FieldWrapSerializerT], _FieldWrapSerializerT] 

239 | Callable[[_FieldPlainSerializerT], _FieldPlainSerializerT] 

240): 

241 """Decorator that enables custom field serialization. 

242 

243 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. 

244 

245 ```python 

246 from typing import Set 

247 

248 from pydantic import BaseModel, field_serializer 

249 

250 class StudentModel(BaseModel): 

251 name: str = 'Jane' 

252 courses: Set[str] 

253 

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

255 def serialize_courses_in_order(self, courses: Set[str]): 

256 return sorted(courses) 

257 

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

259 print(student.model_dump_json()) 

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

261 ``` 

262 

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

264 

265 Four signatures are supported: 

266 

267 - `(self, value: Any, info: FieldSerializationInfo)` 

268 - `(self, value: Any, nxt: SerializerFunctionWrapHandler, info: FieldSerializationInfo)` 

269 - `(value: Any, info: SerializationInfo)` 

270 - `(value: Any, nxt: SerializerFunctionWrapHandler, info: SerializationInfo)` 

271 

272 Args: 

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

274 mode: The serialization mode. 

275 

276 - `plain` means the function will be called instead of the default serialization logic, 

277 - `wrap` means the function will be called with an argument to optionally call the 

278 default serialization logic. 

279 return_type: Optional return type for the function, if omitted it will be inferred from the type annotation. 

280 when_used: Determines the serializer will be used for serialization. 

281 check_fields: Whether to check that the fields actually exist on the model. 

282 

283 Returns: 

284 The decorator function. 

285 """ 

286 

287 def dec(f: FieldSerializer) -> _decorators.PydanticDescriptorProxy[Any]: 

288 dec_info = _decorators.FieldSerializerDecoratorInfo( 

289 fields=fields, 

290 mode=mode, 

291 return_type=return_type, 

292 when_used=when_used, 

293 check_fields=check_fields, 

294 ) 

295 return _decorators.PydanticDescriptorProxy(f, dec_info) # pyright: ignore[reportArgumentType] 

296 

297 return dec # pyright: ignore[reportReturnType] 

298 

299 

300if TYPE_CHECKING: 

301 # The first argument in the following callables represent the `self` type: 

302 

303 ModelPlainSerializerWithInfo: TypeAlias = Callable[[Any, SerializationInfo], Any] 

304 """A model serializer method with the `info` argument, in `plain` mode.""" 

305 

306 ModelPlainSerializerWithoutInfo: TypeAlias = Callable[[Any], Any] 

307 """A model serializer method without the `info` argument, in `plain` mode.""" 

308 

309 ModelPlainSerializer: TypeAlias = 'ModelPlainSerializerWithInfo | ModelPlainSerializerWithoutInfo' 

310 """A model serializer method in `plain` mode.""" 

311 

312 ModelWrapSerializerWithInfo: TypeAlias = Callable[[Any, SerializerFunctionWrapHandler, SerializationInfo], Any] 

313 """A model serializer method with the `info` argument, in `wrap` mode.""" 

314 

315 ModelWrapSerializerWithoutInfo: TypeAlias = Callable[[Any, SerializerFunctionWrapHandler], Any] 

316 """A model serializer method without the `info` argument, in `wrap` mode.""" 

317 

318 ModelWrapSerializer: TypeAlias = 'ModelWrapSerializerWithInfo | ModelWrapSerializerWithoutInfo' 

319 """A model serializer method in `wrap` mode.""" 

320 

321 ModelSerializer: TypeAlias = 'ModelPlainSerializer | ModelWrapSerializer' 

322 

323 _ModelPlainSerializerT = TypeVar('_ModelPlainSerializerT', bound=ModelPlainSerializer) 

324 _ModelWrapSerializerT = TypeVar('_ModelWrapSerializerT', bound=ModelWrapSerializer) 

325 

326 

327@overload 

328def model_serializer(f: _ModelPlainSerializerT, /) -> _ModelPlainSerializerT: ... 

329 

330 

331@overload 

332def model_serializer( 

333 *, mode: Literal['wrap'], when_used: WhenUsed = 'always', return_type: Any = ... 

334) -> Callable[[_ModelWrapSerializerT], _ModelWrapSerializerT]: ... 

335 

336 

337@overload 

338def model_serializer( 

339 *, 

340 mode: Literal['plain'] = ..., 

341 when_used: WhenUsed = 'always', 

342 return_type: Any = ..., 

343) -> Callable[[_ModelPlainSerializerT], _ModelPlainSerializerT]: ... 

344 

345 

346def model_serializer( 

347 f: _ModelPlainSerializerT | _ModelWrapSerializerT | None = None, 

348 /, 

349 *, 

350 mode: Literal['plain', 'wrap'] = 'plain', 

351 when_used: WhenUsed = 'always', 

352 return_type: Any = PydanticUndefined, 

353) -> ( 

354 _ModelPlainSerializerT 

355 | Callable[[_ModelWrapSerializerT], _ModelWrapSerializerT] 

356 | Callable[[_ModelPlainSerializerT], _ModelPlainSerializerT] 

357): 

358 """Decorator that enables custom model serialization. 

359 

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

361 

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

363 

364 ```python 

365 from typing import Literal 

366 

367 from pydantic import BaseModel, model_serializer 

368 

369 class TemperatureModel(BaseModel): 

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

371 value: int 

372 

373 @model_serializer() 

374 def serialize_model(self): 

375 if self.unit == 'F': 

376 return {'unit': 'C', 'value': int((self.value - 32) / 1.8)} 

377 return {'unit': self.unit, 'value': self.value} 

378 

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

380 print(temperature.model_dump()) 

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

382 ``` 

383 

384 Two signatures are supported for `mode='plain'`, which is the default: 

385 

386 - `(self)` 

387 - `(self, info: SerializationInfo)` 

388 

389 And two other signatures for `mode='wrap'`: 

390 

391 - `(self, nxt: SerializerFunctionWrapHandler)` 

392 - `(self, nxt: SerializerFunctionWrapHandler, info: SerializationInfo)` 

393 

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

395 

396 Args: 

397 f: The function to be decorated. 

398 mode: The serialization mode. 

399 

400 - `'plain'` means the function will be called instead of the default serialization logic 

401 - `'wrap'` means the function will be called with an argument to optionally call the default 

402 serialization logic. 

403 when_used: Determines when this serializer should be used. 

404 return_type: The return type for the function. If omitted it will be inferred from the type annotation. 

405 

406 Returns: 

407 The decorator function. 

408 """ 

409 

410 def dec(f: ModelSerializer) -> _decorators.PydanticDescriptorProxy[Any]: 

411 dec_info = _decorators.ModelSerializerDecoratorInfo(mode=mode, return_type=return_type, when_used=when_used) 

412 return _decorators.PydanticDescriptorProxy(f, dec_info) 

413 

414 if f is None: 

415 return dec # pyright: ignore[reportReturnType] 

416 else: 

417 return dec(f) # pyright: ignore[reportReturnType] 

418 

419 

420AnyType = TypeVar('AnyType') 

421 

422 

423if TYPE_CHECKING: 

424 SerializeAsAny = Annotated[AnyType, ...] # SerializeAsAny[list[str]] will be treated by type checkers as list[str] 

425 """Force serialization to ignore whatever is defined in the schema and instead ask the object 

426 itself how it should be serialized. 

427 In particular, this means that when model subclasses are serialized, fields present in the subclass 

428 but not in the original schema will be included. 

429 """ 

430else: 

431 

432 @dataclasses.dataclass(**_internal_dataclass.slots_true) 

433 class SerializeAsAny: # noqa: D101 

434 def __class_getitem__(cls, item: Any) -> Any: 

435 return Annotated[item, SerializeAsAny()] 

436 

437 def __get_pydantic_core_schema__( 

438 self, source_type: Any, handler: GetCoreSchemaHandler 

439 ) -> core_schema.CoreSchema: 

440 schema = handler(source_type) 

441 schema_to_update = schema 

442 while schema_to_update['type'] == 'definitions': 

443 schema_to_update = schema_to_update.copy() 

444 schema_to_update = schema_to_update['schema'] 

445 schema_to_update['serialization'] = core_schema.wrap_serializer_function_ser_schema( 

446 lambda x, h: h(x), schema=core_schema.any_schema() 

447 ) 

448 return schema 

449 

450 __hash__ = object.__hash__