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"""Logic related to validators applied to models etc. via the `@field_validator` and `@model_validator` decorators."""
3from __future__ import annotations as _annotations
5import types
6from collections import deque
7from collections.abc import Callable, Iterable
8from copy import copy
9from dataclasses import dataclass, field
10from functools import cached_property, partial, partialmethod
11from inspect import Parameter, Signature, isdatadescriptor, ismethoddescriptor
12from itertools import islice
13from typing import TYPE_CHECKING, Any, ClassVar, Generic, Literal, TypeAlias, TypeVar
15from pydantic_core import PydanticUndefined, PydanticUndefinedType, core_schema
16from typing_extensions import Self, TypeForm, is_typeddict
18from ..errors import PydanticUserError
19from . import _type_refs
20from ._namespace_utils import GlobalsNamespace, MappingNamespace
21from ._typing_extra import get_function_type_hints, signature_no_eval
22from ._utils import can_be_positional
24if TYPE_CHECKING:
25 from ..fields import ComputedFieldInfo
26 from ..functional_validators import FieldValidatorModes
27 from ._config import ConfigWrapper
30@dataclass(slots=True)
31class ValidatorDecoratorInfo:
32 """A container for data from `@validator` so that we can access it
33 while building the pydantic-core schema.
35 Attributes:
36 decorator_repr: A class variable representing the decorator string, '@validator'.
37 fields: A tuple of field names the validator should be called on.
38 mode: The proposed validator mode.
39 each_item: For complex objects (sets, lists etc.) whether to validate individual
40 elements rather than the whole object.
41 always: Whether this method and other validators should be called even if the value is missing.
42 check_fields: Whether to check that the fields actually exist on the model.
43 """
45 decorator_repr: ClassVar[str] = '@validator'
47 fields: tuple[str, ...]
48 mode: Literal['before', 'after']
49 each_item: bool
50 always: bool
51 check_fields: bool | None
54@dataclass(slots=True)
55class FieldValidatorDecoratorInfo:
56 """A container for data from `@field_validator` so that we can access it
57 while building the pydantic-core schema.
59 Attributes:
60 decorator_repr: A class variable representing the decorator string, '@field_validator'.
61 fields: A tuple of field names the validator should be called on.
62 mode: The proposed validator mode.
63 check_fields: Whether to check that the fields actually exist on the model.
64 json_schema_input_type: The input type of the function. This is only used to generate
65 the appropriate JSON Schema (in validation mode) and can only specified
66 when `mode` is either `'before'`, `'plain'` or `'wrap'`.
67 """
69 decorator_repr: ClassVar[str] = '@field_validator'
71 fields: tuple[str, ...]
72 mode: FieldValidatorModes
73 check_fields: bool | None
74 json_schema_input_type: Any
77@dataclass(slots=True)
78class RootValidatorDecoratorInfo:
79 """A container for data from `@root_validator` so that we can access it
80 while building the pydantic-core schema.
82 Attributes:
83 decorator_repr: A class variable representing the decorator string, '@root_validator'.
84 mode: The proposed validator mode.
85 """
87 decorator_repr: ClassVar[str] = '@root_validator'
88 mode: Literal['before', 'after']
91@dataclass(slots=True)
92class FieldSerializerDecoratorInfo:
93 """A container for data from `@field_serializer` so that we can access it
94 while building the pydantic-core schema.
96 Attributes:
97 decorator_repr: A class variable representing the decorator string, '@field_serializer'.
98 fields: A tuple of field names the serializer should be called on.
99 mode: The proposed serializer mode.
100 return_type: The type of the serializer's return value.
101 when_used: The serialization condition. Accepts a string with values `'always'`, `'unless-none'`, `'json'`,
102 and `'json-unless-none'`.
103 check_fields: Whether to check that the fields actually exist on the model.
104 """
106 decorator_repr: ClassVar[str] = '@field_serializer'
107 fields: tuple[str, ...]
108 mode: Literal['plain', 'wrap']
109 return_type: TypeForm[Any]
110 when_used: core_schema.WhenUsed
111 check_fields: bool | None
114@dataclass(slots=True)
115class ModelSerializerDecoratorInfo:
116 """A container for data from `@model_serializer` so that we can access it
117 while building the pydantic-core schema.
119 Attributes:
120 decorator_repr: A class variable representing the decorator string, '@model_serializer'.
121 mode: The proposed serializer mode.
122 return_type: The type of the serializer's return value.
123 when_used: The serialization condition. Accepts a string with values `'always'`, `'unless-none'`, `'json'`,
124 and `'json-unless-none'`.
125 """
127 decorator_repr: ClassVar[str] = '@model_serializer'
128 mode: Literal['plain', 'wrap']
129 return_type: TypeForm[Any]
130 when_used: core_schema.WhenUsed
133@dataclass(slots=True)
134class ModelValidatorDecoratorInfo:
135 """A container for data from `@model_validator` so that we can access it
136 while building the pydantic-core schema.
138 Attributes:
139 decorator_repr: A class variable representing the decorator string, '@model_validator'.
140 mode: The proposed serializer mode.
141 """
143 decorator_repr: ClassVar[str] = '@model_validator'
144 mode: Literal['wrap', 'before', 'after']
147DecoratorInfo: TypeAlias = 'ValidatorDecoratorInfo | FieldValidatorDecoratorInfo | RootValidatorDecoratorInfo | FieldSerializerDecoratorInfo | ModelSerializerDecoratorInfo | ModelValidatorDecoratorInfo | ComputedFieldInfo'
149ReturnType = TypeVar('ReturnType')
150DecoratedType: TypeAlias = (
151 'classmethod[Any, Any, ReturnType] | staticmethod[Any, ReturnType] | Callable[..., ReturnType] | property'
152)
155@dataclass # can't use slots here since we set attributes on `__post_init__`
156class PydanticDescriptorProxy(Generic[ReturnType]):
157 """Wrap a classmethod, staticmethod, property or unbound function
158 and act as a descriptor that allows us to detect decorated items
159 from the class' attributes.
161 This class' __get__ returns the wrapped item's __get__ result,
162 which makes it transparent for classmethods and staticmethods.
164 Attributes:
165 wrapped: The decorator that has to be wrapped.
166 decorator_info: The decorator info.
167 shim: A wrapper function to wrap V1 style function.
168 """
170 wrapped: DecoratedType[ReturnType]
171 decorator_info: DecoratorInfo
172 shim: Callable[[Callable[..., Any]], Callable[..., Any]] | None = None
174 def __post_init__(self):
175 for attr in 'setter', 'deleter':
176 if hasattr(self.wrapped, attr):
177 f = partial(self._call_wrapped_attr, name=attr)
178 setattr(self, attr, f)
180 def _call_wrapped_attr(self, func: Callable[[Any], None], *, name: str) -> PydanticDescriptorProxy[ReturnType]:
181 self.wrapped = getattr(self.wrapped, name)(func)
182 if isinstance(self.wrapped, property):
183 # update ComputedFieldInfo.wrapped_property
184 from ..fields import ComputedFieldInfo
186 if isinstance(self.decorator_info, ComputedFieldInfo):
187 self.decorator_info.wrapped_property = self.wrapped
188 return self
190 def __get__(self, obj: object | None, obj_type: type[object] | None = None) -> PydanticDescriptorProxy[ReturnType]:
191 try:
192 return self.wrapped.__get__(obj, obj_type) # pyright: ignore[reportReturnType]
193 except AttributeError:
194 # not a descriptor, e.g. a partial object
195 return self.wrapped # type: ignore[return-value]
197 def __set_name__(self, instance: Any, name: str) -> None:
198 if hasattr(self.wrapped, '__set_name__'):
199 self.wrapped.__set_name__(instance, name) # pyright: ignore[reportFunctionMemberAccess]
201 def __getattr__(self, name: str, /) -> Any:
202 """Forward checks for __isabstractmethod__ and such."""
203 return getattr(self.wrapped, name)
206DecoratorInfoType = TypeVar('DecoratorInfoType', bound=DecoratorInfo)
209@dataclass(slots=True)
210class Decorator(Generic[DecoratorInfoType]):
211 """A generic container class to join together the decorator metadata
212 (metadata from decorator itself, which we have when the
213 decorator is called but not when we are building the core-schema)
214 and the bound function (which we have after the class itself is created).
216 Attributes:
217 cls_ref: The class ref.
218 cls_var_name: The decorated function name.
219 func: The decorated function.
220 shim: A wrapper function to wrap V1 style function.
221 info: The decorator info.
222 """
224 cls_ref: str
225 cls_var_name: str
226 func: Callable[..., Any]
227 shim: Callable[[Any], Any] | None
228 info: DecoratorInfoType
230 @staticmethod
231 def build(
232 cls_: Any,
233 *,
234 cls_var_name: str,
235 cls_ref: str | None = None,
236 shim: Callable[[Any], Any] | None,
237 info: DecoratorInfoType,
238 ) -> Decorator[DecoratorInfoType]:
239 """Build a new decorator.
241 Args:
242 cls_: The class.
243 cls_ref: The type ref of the class.
244 cls_var_name: The decorated function name.
245 shim: A wrapper function to wrap V1 style function.
246 info: The decorator info.
248 Returns:
249 The new decorator instance.
250 """
251 # For backwards compatibility for users using this (private!) method. TODO v3: make `cls_ref` required:
252 cls_ref = cls_ref if cls_ref is not None else _type_refs.any_class_type_ref(cls_)
254 func = get_attribute_from_bases(cls_, cls_var_name)
255 if shim is not None:
256 func = shim(func)
257 func = unwrap_wrapped_function(func, unwrap_partial=False)
258 if not callable(func):
259 # TODO most likely this branch can be removed when we drop support for Python 3.12:
260 # This branch will get hit for classmethod properties
261 attribute = get_attribute_from_base_dicts(cls_, cls_var_name) # prevents the binding call to `__get__`
262 if isinstance(attribute, PydanticDescriptorProxy):
263 func = unwrap_wrapped_function(attribute.wrapped)
264 return Decorator(
265 cls_ref=cls_ref,
266 cls_var_name=cls_var_name,
267 func=func,
268 shim=shim,
269 info=info,
270 )
272 def bind_to_cls(self, cls: Any, cls_ref: str) -> Decorator[DecoratorInfoType]:
273 """Bind the decorator to a class.
275 Args:
276 cls: the class.
277 cls_ref: The type ref of the class.
279 Returns:
280 The new decorator instance.
281 """
282 return self.build(
283 cls,
284 cls_ref=cls_ref,
285 cls_var_name=self.cls_var_name,
286 shim=self.shim,
287 info=copy(self.info),
288 )
291def get_bases(tp: type[Any]) -> tuple[type[Any], ...]:
292 """Get the base classes of a class or typeddict.
294 Args:
295 tp: The type or class to get the bases.
297 Returns:
298 The base classes.
299 """
300 if is_typeddict(tp):
301 return tp.__orig_bases__ # type: ignore
302 try:
303 return tp.__bases__
304 except AttributeError:
305 return ()
308def mro(tp: type[Any]) -> tuple[type[Any], ...]:
309 """Calculate the Method Resolution Order of bases using the C3 algorithm.
311 See https://www.python.org/download/releases/2.3/mro/
312 """
313 # try to use the existing mro, for performance mainly
314 # but also because it helps verify the implementation below
315 if not is_typeddict(tp):
316 try:
317 return tp.__mro__
318 except AttributeError:
319 # GenericAlias and some other cases
320 pass
322 bases = get_bases(tp)
323 return (tp,) + mro_for_bases(bases)
326def mro_for_bases(bases: tuple[type[Any], ...]) -> tuple[type[Any], ...]:
327 def merge_seqs(seqs: list[deque[type[Any]]]) -> Iterable[type[Any]]:
328 while True:
329 non_empty = [seq for seq in seqs if seq]
330 if not non_empty:
331 # Nothing left to process, we're done.
332 return
333 candidate: type[Any] | None = None
334 for seq in non_empty: # Find merge candidates among seq heads.
335 candidate = seq[0]
336 not_head = [s for s in non_empty if candidate in islice(s, 1, None)]
337 if not_head:
338 # Reject the candidate.
339 candidate = None
340 else:
341 break
342 if not candidate:
343 raise TypeError('Inconsistent hierarchy, no C3 MRO is possible')
344 yield candidate
345 for seq in non_empty:
346 # Remove candidate.
347 if seq[0] == candidate:
348 seq.popleft()
350 seqs = [deque(mro(base)) for base in bases] + [deque(bases)]
351 return tuple(merge_seqs(seqs))
354_sentinel = object()
357def get_attribute_from_bases(tp: type[Any] | tuple[type[Any], ...], name: str) -> Any:
358 """Get the attribute from the next class in the MRO that has it,
359 aiming to simulate calling the method on the actual class.
361 The reason for iterating over the mro instead of just getting
362 the attribute (which would do that for us) is to support TypedDict,
363 which lacks a real __mro__, but can have a virtual one constructed
364 from its bases (as done here).
366 Args:
367 tp: The type or class to search for the attribute. If a tuple, this is treated as a set of base classes.
368 name: The name of the attribute to retrieve.
370 Returns:
371 Any: The attribute value, if found.
373 Raises:
374 AttributeError: If the attribute is not found in any class in the MRO.
375 """
376 if isinstance(tp, tuple):
377 for base in mro_for_bases(tp):
378 attribute = base.__dict__.get(name, _sentinel)
379 if attribute is not _sentinel:
380 attribute_get = getattr(attribute, '__get__', None)
381 if attribute_get is not None:
382 return attribute_get(None, tp)
383 return attribute
384 raise AttributeError(f'{name} not found in {tp}')
385 else:
386 try:
387 return getattr(tp, name)
388 except AttributeError:
389 return get_attribute_from_bases(mro(tp), name)
392def get_attribute_from_base_dicts(tp: type[Any], name: str) -> Any:
393 """Get an attribute out of the `__dict__` following the MRO.
394 This prevents the call to `__get__` on the descriptor, and allows
395 us to get the original function for classmethod properties.
397 Args:
398 tp: The type or class to search for the attribute.
399 name: The name of the attribute to retrieve.
401 Returns:
402 Any: The attribute value, if found.
404 Raises:
405 KeyError: If the attribute is not found in any class's `__dict__` in the MRO.
406 """
407 for base in reversed(mro(tp)):
408 if name in base.__dict__:
409 return base.__dict__[name]
410 return tp.__dict__[name] # raise the error
413@dataclass(slots=True)
414class DecoratorInfos:
415 """Mapping of name in the class namespace to decorator info.
417 note that the name in the class namespace is the function or attribute name
418 not the field name!
419 """
421 validators: dict[str, Decorator[ValidatorDecoratorInfo]] = field(default_factory=dict)
422 field_validators: dict[str, Decorator[FieldValidatorDecoratorInfo]] = field(default_factory=dict)
423 root_validators: dict[str, Decorator[RootValidatorDecoratorInfo]] = field(default_factory=dict)
424 field_serializers: dict[str, Decorator[FieldSerializerDecoratorInfo]] = field(default_factory=dict)
425 model_serializers: dict[str, Decorator[ModelSerializerDecoratorInfo]] = field(default_factory=dict)
426 model_validators: dict[str, Decorator[ModelValidatorDecoratorInfo]] = field(default_factory=dict)
427 computed_fields: dict[str, Decorator[ComputedFieldInfo]] = field(default_factory=dict)
429 @classmethod
430 def build(
431 cls,
432 typ: type[Any],
433 # Default to `True` for backwards compatibility:
434 replace_wrapped_methods: bool = True,
435 ) -> Self:
436 """Build a `DecoratorInfos` instance for the given model, dataclass or `TypedDict` type.
438 Decorators from parent classes are included, including "bare" classes (e.g. if `typ`
439 is a Pydantic model, non Pydantic parent model classes are also taken into account).
440 The collection of the decorators happens by respecting the MRO.
442 If one of the bases has an `__pydantic_decorators__` attribute set, it is assumed to be
443 a `DecoratorInfos` instance and is used as-is. The `__pydantic_decorators__` attribute
444 is *not* being set on the provided `typ`.
446 Args:
447 typ: The model, dataclass or `TypedDict` type to use when building the `DecoratorInfos` instance.
448 replace_wrapped_methods: Whether to replace the decorator's wrapped methods on `typ`.
449 This is useful e.g. for field validators which are initially class methods. This should
450 only be set to `True` if `typ` is a Pydantic model or dataclass (otherwise this results
451 in mutations of classes Pydantic doesn't "own").
452 """
453 # reminder: dicts are ordered and replacement does not alter the order
454 res = cls()
456 type_ref: str | None = None
458 def cls_ref() -> str:
459 # The type ref is only required if at least one decorator is collected,
460 # so only compute it lazily (and only once):
461 nonlocal type_ref
462 if type_ref is None:
463 type_ref = _type_refs.any_class_type_ref(typ)
464 return type_ref
466 # Iterate over the bases, without the actual `typ`.
467 # `1:-1` because we don't need to include `object`/`TypedDict`:
468 for base in reversed(mro(typ)[1:-1]):
469 existing: DecoratorInfos | None = base.__dict__.get('__pydantic_decorators__')
470 if existing is None:
471 existing, _ = _decorator_infos_for_class(
472 base,
473 # The class refs of the built decorators are irrelevant here, as `bind_to_cls()`
474 # below rebinds every decorator to `typ` with the correct ref:
475 cls_ref=lambda: '',
476 collect_to_replace=False,
477 )
478 res.validators.update({k: v.bind_to_cls(typ, cls_ref()) for k, v in existing.validators.items()})
479 res.field_validators.update(
480 {k: v.bind_to_cls(typ, cls_ref()) for k, v in existing.field_validators.items()}
481 )
482 res.root_validators.update({k: v.bind_to_cls(typ, cls_ref()) for k, v in existing.root_validators.items()})
483 res.field_serializers.update(
484 {k: v.bind_to_cls(typ, cls_ref()) for k, v in existing.field_serializers.items()}
485 )
486 res.model_serializers.update(
487 {k: v.bind_to_cls(typ, cls_ref()) for k, v in existing.model_serializers.items()}
488 )
489 res.model_validators.update(
490 {k: v.bind_to_cls(typ, cls_ref()) for k, v in existing.model_validators.items()}
491 )
492 res.computed_fields.update({k: v.bind_to_cls(typ, cls_ref()) for k, v in existing.computed_fields.items()})
494 decorator_infos, to_replace = _decorator_infos_for_class(typ, cls_ref=cls_ref, collect_to_replace=True)
496 res.validators.update(decorator_infos.validators)
497 res.field_validators.update(decorator_infos.field_validators)
498 res.root_validators.update(decorator_infos.root_validators)
499 res.field_serializers.update(decorator_infos.field_serializers)
500 res.model_serializers.update(decorator_infos.model_serializers)
501 res.model_validators.update(decorator_infos.model_validators)
502 res.computed_fields.update(decorator_infos.computed_fields)
504 if replace_wrapped_methods and to_replace:
505 for name, value in to_replace:
506 setattr(typ, name, value)
508 res._validate()
509 return res
511 def _validate(self) -> None:
512 seen: set[str] = set()
513 for field_ser in self.field_serializers.values():
514 for f_name in field_ser.info.fields:
515 if f_name in seen:
516 raise PydanticUserError(
517 f'Multiple field serializer functions were defined for field {f_name!r}, this is not allowed.',
518 code='multiple-field-serializers',
519 )
520 seen.add(f_name)
522 def update_from_config(self, config_wrapper: ConfigWrapper) -> None:
523 """Update the decorator infos from the configuration of the class they are attached to."""
524 for name, computed_field_dec in self.computed_fields.items():
525 computed_field_dec.info._update_from_config(config_wrapper, name)
528def _decorator_infos_for_class(
529 typ: type[Any],
530 *,
531 cls_ref: Callable[[], str],
532 collect_to_replace: bool,
533) -> tuple[DecoratorInfos, list[tuple[str, Any]]]:
534 """Collect a `DecoratorInfos` for class, without looking into bases."""
535 res = DecoratorInfos()
536 to_replace: list[tuple[str, Any]] = []
538 # Iterate on the vars with a copy to avoid changes in dictionary size during iteration.
539 # This can happen if a concurrent thread has Pydantic processing the same `typ` (e.g. two
540 # concurrent threads doing `TypeAdapter(SomeClassType)`), and such thread reads annotations
541 # for the first time during the iteration here in the other thread (since 3.14, reading annotations
542 # can set attributes such as `__annotations_cache__` or `__annotate_func__` to the type).
543 for var_name, var_value in vars(typ).copy().items():
544 if isinstance(var_value, PydanticDescriptorProxy):
545 info = var_value.decorator_info
546 if isinstance(info, ValidatorDecoratorInfo):
547 res.validators[var_name] = Decorator.build(
548 typ, cls_ref=cls_ref(), cls_var_name=var_name, shim=var_value.shim, info=info
549 )
550 elif isinstance(info, FieldValidatorDecoratorInfo):
551 res.field_validators[var_name] = Decorator.build(
552 typ, cls_ref=cls_ref(), cls_var_name=var_name, shim=var_value.shim, info=info
553 )
554 elif isinstance(info, RootValidatorDecoratorInfo):
555 res.root_validators[var_name] = Decorator.build(
556 typ, cls_ref=cls_ref(), cls_var_name=var_name, shim=var_value.shim, info=info
557 )
558 elif isinstance(info, FieldSerializerDecoratorInfo):
559 res.field_serializers[var_name] = Decorator.build(
560 typ, cls_ref=cls_ref(), cls_var_name=var_name, shim=var_value.shim, info=info
561 )
562 elif isinstance(info, ModelValidatorDecoratorInfo):
563 res.model_validators[var_name] = Decorator.build(
564 typ, cls_ref=cls_ref(), cls_var_name=var_name, shim=var_value.shim, info=info
565 )
566 elif isinstance(info, ModelSerializerDecoratorInfo):
567 res.model_serializers[var_name] = Decorator.build(
568 typ, cls_ref=cls_ref(), cls_var_name=var_name, shim=var_value.shim, info=info
569 )
570 else:
571 from ..fields import ComputedFieldInfo
573 assert isinstance(info, ComputedFieldInfo)
574 res.computed_fields[var_name] = Decorator.build(
575 typ, cls_ref=cls_ref(), cls_var_name=var_name, shim=None, info=info
576 )
577 if collect_to_replace:
578 to_replace.append((var_name, var_value.wrapped))
580 return res, to_replace
583def inspect_validator(
584 validator: Callable[..., Any], *, mode: FieldValidatorModes, type: Literal['field', 'model']
585) -> bool:
586 """Look at a field or model validator function and determine whether it takes an info argument.
588 An error is raised if the function has an invalid signature.
590 Args:
591 validator: The validator function to inspect.
592 mode: The proposed validator mode.
593 type: The type of validator, either 'field' or 'model'.
595 Returns:
596 Whether the validator takes an info argument.
597 """
598 try:
599 sig = signature_no_eval(validator)
600 except (ValueError, TypeError):
601 # `inspect.signature` might not be able to infer a signature, e.g. with C objects.
602 # In this case, we assume no info argument is present:
603 return False
604 n_positional = count_positional_required_params(sig)
605 if mode == 'wrap':
606 if n_positional == 3:
607 return True
608 elif n_positional == 2:
609 return False
610 else:
611 assert mode in {'before', 'after', 'plain'}, f"invalid mode: {mode!r}, expected 'before', 'after' or 'plain"
612 if n_positional == 2:
613 return True
614 elif n_positional == 1:
615 return False
617 raise PydanticUserError(
618 f'Unrecognized {type} validator function signature for {validator} with `mode={mode}`: {sig}',
619 code='validator-signature',
620 )
623def inspect_field_serializer(serializer: Callable[..., Any], mode: Literal['plain', 'wrap']) -> tuple[bool, bool]:
624 """Look at a field serializer function and determine if it is a field serializer,
625 and whether it takes an info argument.
627 An error is raised if the function has an invalid signature.
629 Args:
630 serializer: The serializer function to inspect.
631 mode: The serializer mode, either 'plain' or 'wrap'.
633 Returns:
634 Tuple of (is_field_serializer, info_arg).
635 """
636 try:
637 sig = signature_no_eval(serializer)
638 except (ValueError, TypeError):
639 # `inspect.signature` might not be able to infer a signature, e.g. with C objects.
640 # In this case, we assume no info argument is present and this is not a method:
641 return (False, False)
643 first = next(iter(sig.parameters.values()), None)
644 is_field_serializer = first is not None and first.name == 'self'
646 n_positional = count_positional_required_params(sig)
647 if is_field_serializer:
648 # -1 to correct for self parameter
649 info_arg = _serializer_info_arg(mode, n_positional - 1)
650 else:
651 info_arg = _serializer_info_arg(mode, n_positional)
653 if info_arg is None:
654 raise PydanticUserError(
655 f'Unrecognized field_serializer function signature for {serializer} with `mode={mode}`:{sig}',
656 code='field-serializer-signature',
657 )
659 return is_field_serializer, info_arg
662def inspect_annotated_serializer(serializer: Callable[..., Any], mode: Literal['plain', 'wrap']) -> bool:
663 """Look at a serializer function used via `Annotated` and determine whether it takes an info argument.
665 An error is raised if the function has an invalid signature.
667 Args:
668 serializer: The serializer function to check.
669 mode: The serializer mode, either 'plain' or 'wrap'.
671 Returns:
672 info_arg
673 """
674 try:
675 sig = signature_no_eval(serializer)
676 except (ValueError, TypeError):
677 # `inspect.signature` might not be able to infer a signature, e.g. with C objects.
678 # In this case, we assume no info argument is present:
679 return False
680 info_arg = _serializer_info_arg(mode, count_positional_required_params(sig))
681 if info_arg is None:
682 raise PydanticUserError(
683 f'Unrecognized field_serializer function signature for {serializer} with `mode={mode}`:{sig}',
684 code='field-serializer-signature',
685 )
686 else:
687 return info_arg
690def inspect_model_serializer(serializer: Callable[..., Any], mode: Literal['plain', 'wrap']) -> bool:
691 """Look at a model serializer function and determine whether it takes an info argument.
693 An error is raised if the function has an invalid signature.
695 Args:
696 serializer: The serializer function to check.
697 mode: The serializer mode, either 'plain' or 'wrap'.
699 Returns:
700 `info_arg` - whether the function expects an info argument.
701 """
702 if isinstance(serializer, (staticmethod, classmethod)) or not is_instance_method_from_sig(serializer):
703 raise PydanticUserError(
704 '`@model_serializer` must be applied to instance methods', code='model-serializer-instance-method'
705 )
707 sig = signature_no_eval(serializer)
708 info_arg = _serializer_info_arg(mode, count_positional_required_params(sig))
709 if info_arg is None:
710 raise PydanticUserError(
711 f'Unrecognized model_serializer function signature for {serializer} with `mode={mode}`:{sig}',
712 code='model-serializer-signature',
713 )
714 else:
715 return info_arg
718def _serializer_info_arg(mode: Literal['plain', 'wrap'], n_positional: int) -> bool | None:
719 if mode == 'plain':
720 if n_positional == 1:
721 # (input_value: Any, /) -> Any
722 return False
723 elif n_positional == 2:
724 # (model: Any, input_value: Any, /) -> Any
725 return True
726 else:
727 assert mode == 'wrap', f"invalid mode: {mode!r}, expected 'plain' or 'wrap'"
728 if n_positional == 2:
729 # (input_value: Any, serializer: SerializerFunctionWrapHandler, /) -> Any
730 return False
731 elif n_positional == 3:
732 # (input_value: Any, serializer: SerializerFunctionWrapHandler, info: SerializationInfo, /) -> Any
733 return True
735 return None
738AnyDecoratorCallable: TypeAlias = (
739 'classmethod[Any, Any, Any] | staticmethod[Any, Any] | partialmethod[Any] | Callable[..., Any]'
740)
743def is_instance_method_from_sig(function: AnyDecoratorCallable) -> bool:
744 """Whether the function is an instance method.
746 It will consider a function as instance method if the first parameter of
747 function is `self`.
749 Args:
750 function: The function to check.
752 Returns:
753 `True` if the function is an instance method, `False` otherwise.
754 """
755 sig = signature_no_eval(unwrap_wrapped_function(function))
756 first = next(iter(sig.parameters.values()), None)
757 if first and first.name == 'self':
758 return True
759 return False
762def ensure_classmethod_based_on_signature(function: AnyDecoratorCallable) -> Any:
763 """Apply the `@classmethod` decorator on the function.
765 Args:
766 function: The function to apply the decorator on.
768 Return:
769 The `@classmethod` decorator applied function.
770 """
771 if not isinstance(
772 unwrap_wrapped_function(function, unwrap_class_static_method=False), classmethod
773 ) and _is_classmethod_from_sig(function):
774 return classmethod(function) # type: ignore[arg-type]
775 return function
778def _is_classmethod_from_sig(function: AnyDecoratorCallable) -> bool:
779 sig = signature_no_eval(unwrap_wrapped_function(function))
780 first = next(iter(sig.parameters.values()), None)
781 if first and first.name == 'cls':
782 return True
783 return False
786def unwrap_wrapped_function(
787 func: Any,
788 *,
789 unwrap_partial: bool = True,
790 unwrap_class_static_method: bool = True,
791) -> Any:
792 """Recursively unwraps a wrapped function until the underlying function is reached.
793 This handles property, functools.partial, functools.partialmethod, staticmethod, and classmethod.
795 Args:
796 func: The function to unwrap.
797 unwrap_partial: If True (default), unwrap partial and partialmethod decorators.
798 unwrap_class_static_method: If True (default), also unwrap classmethod and staticmethod
799 decorators. If False, only unwrap partial and partialmethod decorators.
801 Returns:
802 The underlying function of the wrapped function.
803 """
804 # Define the types we want to check against as a single tuple.
805 unwrap_types = (
806 (property, cached_property)
807 + ((partial, partialmethod) if unwrap_partial else ())
808 + ((staticmethod, classmethod) if unwrap_class_static_method else ())
809 )
811 while isinstance(func, unwrap_types):
812 if unwrap_class_static_method and isinstance(func, (classmethod, staticmethod)):
813 func = func.__func__
814 elif isinstance(func, (partial, partialmethod)):
815 func = func.func
816 elif isinstance(func, property):
817 func = func.fget # arbitrary choice, convenient for computed fields
818 else:
819 # Make coverage happy as it can only get here in the last possible case
820 assert isinstance(func, cached_property)
821 func = func.func # type: ignore
823 return func
826_function_like = (
827 partial,
828 partialmethod,
829 types.FunctionType,
830 types.BuiltinFunctionType,
831 types.MethodType,
832 types.WrapperDescriptorType,
833 types.MethodWrapperType,
834 types.MemberDescriptorType,
835)
838def get_callable_return_type(
839 callable_obj: Any,
840 globalns: GlobalsNamespace | None = None,
841 localns: MappingNamespace | None = None,
842) -> Any | PydanticUndefinedType:
843 """Get the callable return type.
845 Args:
846 callable_obj: The callable to analyze.
847 globalns: The globals namespace to use during type annotation evaluation.
848 localns: The locals namespace to use during type annotation evaluation.
850 Returns:
851 The function return type.
852 """
853 if isinstance(callable_obj, type):
854 # types are callables, and we assume the return type
855 # is the type itself (e.g. `int()` results in an instance of `int`).
856 return callable_obj
858 if not isinstance(callable_obj, _function_like):
859 call_func = getattr(type(callable_obj), '__call__', None) # noqa: B004
860 if call_func is not None:
861 callable_obj = call_func
863 hints = get_function_type_hints(
864 unwrap_wrapped_function(callable_obj),
865 include_keys={'return'},
866 globalns=globalns,
867 localns=localns,
868 )
869 return hints.get('return', PydanticUndefined)
872def count_positional_required_params(sig: Signature) -> int:
873 """Get the number of positional (required) arguments of a signature.
875 This function should only be used to inspect signatures of validation and serialization functions.
876 The first argument (the value being serialized or validated) is counted as a required argument
877 even if a default value exists.
879 Returns:
880 The number of positional arguments of a signature.
881 """
882 parameters = list(sig.parameters.values())
883 return sum(
884 1
885 for param in parameters
886 if can_be_positional(param)
887 # First argument is the value being validated/serialized, and can have a default value
888 # (e.g. `float`, which has signature `(x=0, /)`). We assume other parameters (the info arg
889 # for instance) should be required, and thus without any default value.
890 and (param.default is Parameter.empty or param is parameters[0])
891 )
894def ensure_property(f: Any) -> Any:
895 """Ensure that a function is a `property` or `cached_property`, or is a valid descriptor.
897 Args:
898 f: The function to check.
900 Returns:
901 The function, or a `property` or `cached_property` instance wrapping the function.
902 """
903 if ismethoddescriptor(f) or isdatadescriptor(f):
904 return f
905 else:
906 return property(f)