| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792 |
- """Private logic for creating models."""
- from __future__ import annotations as _annotations
- import builtins
- import operator
- import sys
- import typing
- import warnings
- import weakref
- from abc import ABCMeta
- from functools import lru_cache, partial
- from types import FunctionType
- from typing import Any, Callable, Generic, Literal, NoReturn, cast
- from pydantic_core import PydanticUndefined, SchemaSerializer
- from typing_extensions import TypeAliasType, dataclass_transform, deprecated, get_args
- from ..errors import PydanticUndefinedAnnotation, PydanticUserError
- from ..plugin._schema_validator import create_schema_validator
- from ..warnings import GenericBeforeBaseModelWarning, PydanticDeprecatedSince20
- from ._config import ConfigWrapper
- from ._decorators import DecoratorInfos, PydanticDescriptorProxy, get_attribute_from_bases, unwrap_wrapped_function
- from ._fields import collect_model_fields, is_valid_field_name, is_valid_privateattr_name
- from ._generate_schema import GenerateSchema
- from ._generics import PydanticGenericMetadata, get_model_typevars_map
- from ._import_utils import import_cached_base_model, import_cached_field_info
- from ._mock_val_ser import set_model_mocks
- from ._namespace_utils import NsResolver
- from ._schema_generation_shared import CallbackGetCoreSchemaHandler
- from ._signature import generate_pydantic_signature
- from ._typing_extra import (
- _make_forward_ref,
- eval_type_backport,
- is_annotated,
- is_classvar_annotation,
- parent_frame_namespace,
- )
- from ._utils import LazyClassAttribute, SafeGetItemProxy
- if typing.TYPE_CHECKING:
- from ..fields import ComputedFieldInfo, FieldInfo, ModelPrivateAttr
- from ..fields import Field as PydanticModelField
- from ..fields import PrivateAttr as PydanticModelPrivateAttr
- from ..main import BaseModel
- else:
- # See PyCharm issues https://youtrack.jetbrains.com/issue/PY-21915
- # and https://youtrack.jetbrains.com/issue/PY-51428
- DeprecationWarning = PydanticDeprecatedSince20
- PydanticModelField = object()
- PydanticModelPrivateAttr = object()
- object_setattr = object.__setattr__
- class _ModelNamespaceDict(dict):
- """A dictionary subclass that intercepts attribute setting on model classes and
- warns about overriding of decorators.
- """
- def __setitem__(self, k: str, v: object) -> None:
- existing: Any = self.get(k, None)
- if existing and v is not existing and isinstance(existing, PydanticDescriptorProxy):
- warnings.warn(f'`{k}` overrides an existing Pydantic `{existing.decorator_info.decorator_repr}` decorator')
- return super().__setitem__(k, v)
- def NoInitField(
- *,
- init: Literal[False] = False,
- ) -> Any:
- """Only for typing purposes. Used as default value of `__pydantic_fields_set__`,
- `__pydantic_extra__`, `__pydantic_private__`, so they could be ignored when
- synthesizing the `__init__` signature.
- """
- @dataclass_transform(kw_only_default=True, field_specifiers=(PydanticModelField, PydanticModelPrivateAttr, NoInitField))
- class ModelMetaclass(ABCMeta):
- def __new__(
- mcs,
- cls_name: str,
- bases: tuple[type[Any], ...],
- namespace: dict[str, Any],
- __pydantic_generic_metadata__: PydanticGenericMetadata | None = None,
- __pydantic_reset_parent_namespace__: bool = True,
- _create_model_module: str | None = None,
- **kwargs: Any,
- ) -> type:
- """Metaclass for creating Pydantic models.
- Args:
- cls_name: The name of the class to be created.
- bases: The base classes of the class to be created.
- namespace: The attribute dictionary of the class to be created.
- __pydantic_generic_metadata__: Metadata for generic models.
- __pydantic_reset_parent_namespace__: Reset parent namespace.
- _create_model_module: The module of the class to be created, if created by `create_model`.
- **kwargs: Catch-all for any other keyword arguments.
- Returns:
- The new class created by the metaclass.
- """
- # Note `ModelMetaclass` refers to `BaseModel`, but is also used to *create* `BaseModel`, so we rely on the fact
- # that `BaseModel` itself won't have any bases, but any subclass of it will, to determine whether the `__new__`
- # call we're in the middle of is for the `BaseModel` class.
- if bases:
- base_field_names, class_vars, base_private_attributes = mcs._collect_bases_data(bases)
- config_wrapper = ConfigWrapper.for_model(bases, namespace, kwargs)
- namespace['model_config'] = config_wrapper.config_dict
- private_attributes = inspect_namespace(
- namespace, config_wrapper.ignored_types, class_vars, base_field_names
- )
- if private_attributes or base_private_attributes:
- original_model_post_init = get_model_post_init(namespace, bases)
- if original_model_post_init is not None:
- # if there are private_attributes and a model_post_init function, we handle both
- def wrapped_model_post_init(self: BaseModel, context: Any, /) -> None:
- """We need to both initialize private attributes and call the user-defined model_post_init
- method.
- """
- init_private_attributes(self, context)
- original_model_post_init(self, context)
- namespace['model_post_init'] = wrapped_model_post_init
- else:
- namespace['model_post_init'] = init_private_attributes
- namespace['__class_vars__'] = class_vars
- namespace['__private_attributes__'] = {**base_private_attributes, **private_attributes}
- cls = cast('type[BaseModel]', super().__new__(mcs, cls_name, bases, namespace, **kwargs))
- BaseModel_ = import_cached_base_model()
- mro = cls.__mro__
- if Generic in mro and mro.index(Generic) < mro.index(BaseModel_):
- warnings.warn(
- GenericBeforeBaseModelWarning(
- 'Classes should inherit from `BaseModel` before generic classes (e.g. `typing.Generic[T]`) '
- 'for pydantic generics to work properly.'
- ),
- stacklevel=2,
- )
- cls.__pydantic_custom_init__ = not getattr(cls.__init__, '__pydantic_base_init__', False)
- cls.__pydantic_post_init__ = (
- None if cls.model_post_init is BaseModel_.model_post_init else 'model_post_init'
- )
- cls.__pydantic_decorators__ = DecoratorInfos.build(cls)
- # Use the getattr below to grab the __parameters__ from the `typing.Generic` parent class
- if __pydantic_generic_metadata__:
- cls.__pydantic_generic_metadata__ = __pydantic_generic_metadata__
- else:
- parent_parameters = getattr(cls, '__pydantic_generic_metadata__', {}).get('parameters', ())
- parameters = getattr(cls, '__parameters__', None) or parent_parameters
- if parameters and parent_parameters and not all(x in parameters for x in parent_parameters):
- from ..root_model import RootModelRootType
- missing_parameters = tuple(x for x in parameters if x not in parent_parameters)
- if RootModelRootType in parent_parameters and RootModelRootType not in parameters:
- # This is a special case where the user has subclassed `RootModel`, but has not parametrized
- # RootModel with the generic type identifiers being used. Ex:
- # class MyModel(RootModel, Generic[T]):
- # root: T
- # Should instead just be:
- # class MyModel(RootModel[T]):
- # root: T
- parameters_str = ', '.join([x.__name__ for x in missing_parameters])
- error_message = (
- f'{cls.__name__} is a subclass of `RootModel`, but does not include the generic type identifier(s) '
- f'{parameters_str} in its parameters. '
- f'You should parametrize RootModel directly, e.g., `class {cls.__name__}(RootModel[{parameters_str}]): ...`.'
- )
- else:
- combined_parameters = parent_parameters + missing_parameters
- parameters_str = ', '.join([str(x) for x in combined_parameters])
- generic_type_label = f'typing.Generic[{parameters_str}]'
- error_message = (
- f'All parameters must be present on typing.Generic;'
- f' you should inherit from {generic_type_label}.'
- )
- if Generic not in bases: # pragma: no cover
- # We raise an error here not because it is desirable, but because some cases are mishandled.
- # It would be nice to remove this error and still have things behave as expected, it's just
- # challenging because we are using a custom `__class_getitem__` to parametrize generic models,
- # and not returning a typing._GenericAlias from it.
- bases_str = ', '.join([x.__name__ for x in bases] + [generic_type_label])
- error_message += (
- f' Note: `typing.Generic` must go last: `class {cls.__name__}({bases_str}): ...`)'
- )
- raise TypeError(error_message)
- cls.__pydantic_generic_metadata__ = {
- 'origin': None,
- 'args': (),
- 'parameters': parameters,
- }
- cls.__pydantic_complete__ = False # Ensure this specific class gets completed
- # preserve `__set_name__` protocol defined in https://peps.python.org/pep-0487
- # for attributes not in `new_namespace` (e.g. private attributes)
- for name, obj in private_attributes.items():
- obj.__set_name__(cls, name)
- if __pydantic_reset_parent_namespace__:
- cls.__pydantic_parent_namespace__ = build_lenient_weakvaluedict(parent_frame_namespace())
- parent_namespace: dict[str, Any] | None = getattr(cls, '__pydantic_parent_namespace__', None)
- if isinstance(parent_namespace, dict):
- parent_namespace = unpack_lenient_weakvaluedict(parent_namespace)
- ns_resolver = NsResolver(parent_namespace=parent_namespace)
- set_model_fields(cls, bases, config_wrapper, ns_resolver)
- if config_wrapper.frozen and '__hash__' not in namespace:
- set_default_hash_func(cls, bases)
- complete_model_class(
- cls,
- cls_name,
- config_wrapper,
- raise_errors=False,
- ns_resolver=ns_resolver,
- create_model_module=_create_model_module,
- )
- # If this is placed before the complete_model_class call above,
- # the generic computed fields return type is set to PydanticUndefined
- cls.__pydantic_computed_fields__ = {
- k: v.info for k, v in cls.__pydantic_decorators__.computed_fields.items()
- }
- set_deprecated_descriptors(cls)
- # using super(cls, cls) on the next line ensures we only call the parent class's __pydantic_init_subclass__
- # I believe the `type: ignore` is only necessary because mypy doesn't realize that this code branch is
- # only hit for _proper_ subclasses of BaseModel
- super(cls, cls).__pydantic_init_subclass__(**kwargs) # type: ignore[misc]
- return cls
- else:
- # These are instance variables, but have been assigned to `NoInitField` to trick the type checker.
- for instance_slot in '__pydantic_fields_set__', '__pydantic_extra__', '__pydantic_private__':
- namespace.pop(
- instance_slot,
- None, # In case the metaclass is used with a class other than `BaseModel`.
- )
- namespace.get('__annotations__', {}).clear()
- return super().__new__(mcs, cls_name, bases, namespace, **kwargs)
- if not typing.TYPE_CHECKING: # pragma: no branch
- # We put `__getattr__` in a non-TYPE_CHECKING block because otherwise, mypy allows arbitrary attribute access
- def __getattr__(self, item: str) -> Any:
- """This is necessary to keep attribute access working for class attribute access."""
- private_attributes = self.__dict__.get('__private_attributes__')
- if private_attributes and item in private_attributes:
- return private_attributes[item]
- raise AttributeError(item)
- @classmethod
- def __prepare__(cls, *args: Any, **kwargs: Any) -> dict[str, object]:
- return _ModelNamespaceDict()
- def __instancecheck__(self, instance: Any) -> bool:
- """Avoid calling ABC _abc_subclasscheck unless we're pretty sure.
- See #3829 and python/cpython#92810
- """
- return hasattr(instance, '__pydantic_validator__') and super().__instancecheck__(instance)
- @staticmethod
- def _collect_bases_data(bases: tuple[type[Any], ...]) -> tuple[set[str], set[str], dict[str, ModelPrivateAttr]]:
- BaseModel = import_cached_base_model()
- field_names: set[str] = set()
- class_vars: set[str] = set()
- private_attributes: dict[str, ModelPrivateAttr] = {}
- for base in bases:
- if issubclass(base, BaseModel) and base is not BaseModel:
- # model_fields might not be defined yet in the case of generics, so we use getattr here:
- field_names.update(getattr(base, '__pydantic_fields__', {}).keys())
- class_vars.update(base.__class_vars__)
- private_attributes.update(base.__private_attributes__)
- return field_names, class_vars, private_attributes
- @property
- @deprecated('The `__fields__` attribute is deprecated, use `model_fields` instead.', category=None)
- def __fields__(self) -> dict[str, FieldInfo]:
- warnings.warn(
- 'The `__fields__` attribute is deprecated, use `model_fields` instead.',
- PydanticDeprecatedSince20,
- stacklevel=2,
- )
- return self.model_fields
- @property
- def model_fields(self) -> dict[str, FieldInfo]:
- """Get metadata about the fields defined on the model.
- Returns:
- A mapping of field names to [`FieldInfo`][pydantic.fields.FieldInfo] objects.
- """
- return getattr(self, '__pydantic_fields__', {})
- @property
- def model_computed_fields(self) -> dict[str, ComputedFieldInfo]:
- """Get metadata about the computed fields defined on the model.
- Returns:
- A mapping of computed field names to [`ComputedFieldInfo`][pydantic.fields.ComputedFieldInfo] objects.
- """
- return getattr(self, '__pydantic_computed_fields__', {})
- def __dir__(self) -> list[str]:
- attributes = list(super().__dir__())
- if '__fields__' in attributes:
- attributes.remove('__fields__')
- return attributes
- def init_private_attributes(self: BaseModel, context: Any, /) -> None:
- """This function is meant to behave like a BaseModel method to initialise private attributes.
- It takes context as an argument since that's what pydantic-core passes when calling it.
- Args:
- self: The BaseModel instance.
- context: The context.
- """
- if getattr(self, '__pydantic_private__', None) is None:
- pydantic_private = {}
- for name, private_attr in self.__private_attributes__.items():
- default = private_attr.get_default()
- if default is not PydanticUndefined:
- pydantic_private[name] = default
- object_setattr(self, '__pydantic_private__', pydantic_private)
- def get_model_post_init(namespace: dict[str, Any], bases: tuple[type[Any], ...]) -> Callable[..., Any] | None:
- """Get the `model_post_init` method from the namespace or the class bases, or `None` if not defined."""
- if 'model_post_init' in namespace:
- return namespace['model_post_init']
- BaseModel = import_cached_base_model()
- model_post_init = get_attribute_from_bases(bases, 'model_post_init')
- if model_post_init is not BaseModel.model_post_init:
- return model_post_init
- def inspect_namespace( # noqa C901
- namespace: dict[str, Any],
- ignored_types: tuple[type[Any], ...],
- base_class_vars: set[str],
- base_class_fields: set[str],
- ) -> dict[str, ModelPrivateAttr]:
- """Iterate over the namespace and:
- * gather private attributes
- * check for items which look like fields but are not (e.g. have no annotation) and warn.
- Args:
- namespace: The attribute dictionary of the class to be created.
- ignored_types: A tuple of ignore types.
- base_class_vars: A set of base class class variables.
- base_class_fields: A set of base class fields.
- Returns:
- A dict contains private attributes info.
- Raises:
- TypeError: If there is a `__root__` field in model.
- NameError: If private attribute name is invalid.
- PydanticUserError:
- - If a field does not have a type annotation.
- - If a field on base class was overridden by a non-annotated attribute.
- """
- from ..fields import ModelPrivateAttr, PrivateAttr
- FieldInfo = import_cached_field_info()
- all_ignored_types = ignored_types + default_ignored_types()
- private_attributes: dict[str, ModelPrivateAttr] = {}
- raw_annotations = namespace.get('__annotations__', {})
- if '__root__' in raw_annotations or '__root__' in namespace:
- raise TypeError("To define root models, use `pydantic.RootModel` rather than a field called '__root__'")
- ignored_names: set[str] = set()
- for var_name, value in list(namespace.items()):
- if var_name == 'model_config' or var_name == '__pydantic_extra__':
- continue
- elif (
- isinstance(value, type)
- and value.__module__ == namespace['__module__']
- and '__qualname__' in namespace
- and value.__qualname__.startswith(namespace['__qualname__'])
- ):
- # `value` is a nested type defined in this namespace; don't error
- continue
- elif isinstance(value, all_ignored_types) or value.__class__.__module__ == 'functools':
- ignored_names.add(var_name)
- continue
- elif isinstance(value, ModelPrivateAttr):
- if var_name.startswith('__'):
- raise NameError(
- 'Private attributes must not use dunder names;'
- f' use a single underscore prefix instead of {var_name!r}.'
- )
- elif is_valid_field_name(var_name):
- raise NameError(
- 'Private attributes must not use valid field names;'
- f' use sunder names, e.g. {"_" + var_name!r} instead of {var_name!r}.'
- )
- private_attributes[var_name] = value
- del namespace[var_name]
- elif isinstance(value, FieldInfo) and not is_valid_field_name(var_name):
- suggested_name = var_name.lstrip('_') or 'my_field' # don't suggest '' for all-underscore name
- raise NameError(
- f'Fields must not use names with leading underscores;'
- f' e.g., use {suggested_name!r} instead of {var_name!r}.'
- )
- elif var_name.startswith('__'):
- continue
- elif is_valid_privateattr_name(var_name):
- if var_name not in raw_annotations or not is_classvar_annotation(raw_annotations[var_name]):
- private_attributes[var_name] = cast(ModelPrivateAttr, PrivateAttr(default=value))
- del namespace[var_name]
- elif var_name in base_class_vars:
- continue
- elif var_name not in raw_annotations:
- if var_name in base_class_fields:
- raise PydanticUserError(
- f'Field {var_name!r} defined on a base class was overridden by a non-annotated attribute. '
- f'All field definitions, including overrides, require a type annotation.',
- code='model-field-overridden',
- )
- elif isinstance(value, FieldInfo):
- raise PydanticUserError(
- f'Field {var_name!r} requires a type annotation', code='model-field-missing-annotation'
- )
- else:
- raise PydanticUserError(
- f'A non-annotated attribute was detected: `{var_name} = {value!r}`. All model fields require a '
- f'type annotation; if `{var_name}` is not meant to be a field, you may be able to resolve this '
- f"error by annotating it as a `ClassVar` or updating `model_config['ignored_types']`.",
- code='model-field-missing-annotation',
- )
- for ann_name, ann_type in raw_annotations.items():
- if (
- is_valid_privateattr_name(ann_name)
- and ann_name not in private_attributes
- and ann_name not in ignored_names
- # This condition can be a false negative when `ann_type` is stringified,
- # but it is handled in most cases in `set_model_fields`:
- and not is_classvar_annotation(ann_type)
- and ann_type not in all_ignored_types
- and getattr(ann_type, '__module__', None) != 'functools'
- ):
- if isinstance(ann_type, str):
- # Walking up the frames to get the module namespace where the model is defined
- # (as the model class wasn't created yet, we unfortunately can't use `cls.__module__`):
- frame = sys._getframe(2)
- if frame is not None:
- try:
- ann_type = eval_type_backport(
- _make_forward_ref(ann_type, is_argument=False, is_class=True),
- globalns=frame.f_globals,
- localns=frame.f_locals,
- )
- except (NameError, TypeError):
- pass
- if is_annotated(ann_type):
- _, *metadata = get_args(ann_type)
- private_attr = next((v for v in metadata if isinstance(v, ModelPrivateAttr)), None)
- if private_attr is not None:
- private_attributes[ann_name] = private_attr
- continue
- private_attributes[ann_name] = PrivateAttr()
- return private_attributes
- def set_default_hash_func(cls: type[BaseModel], bases: tuple[type[Any], ...]) -> None:
- base_hash_func = get_attribute_from_bases(bases, '__hash__')
- new_hash_func = make_hash_func(cls)
- if base_hash_func in {None, object.__hash__} or getattr(base_hash_func, '__code__', None) == new_hash_func.__code__:
- # If `__hash__` is some default, we generate a hash function.
- # It will be `None` if not overridden from BaseModel.
- # It may be `object.__hash__` if there is another
- # parent class earlier in the bases which doesn't override `__hash__` (e.g. `typing.Generic`).
- # It may be a value set by `set_default_hash_func` if `cls` is a subclass of another frozen model.
- # In the last case we still need a new hash function to account for new `model_fields`.
- cls.__hash__ = new_hash_func
- def make_hash_func(cls: type[BaseModel]) -> Any:
- getter = operator.itemgetter(*cls.__pydantic_fields__.keys()) if cls.__pydantic_fields__ else lambda _: 0
- def hash_func(self: Any) -> int:
- try:
- return hash(getter(self.__dict__))
- except KeyError:
- # In rare cases (such as when using the deprecated copy method), the __dict__ may not contain
- # all model fields, which is how we can get here.
- # getter(self.__dict__) is much faster than any 'safe' method that accounts for missing keys,
- # and wrapping it in a `try` doesn't slow things down much in the common case.
- return hash(getter(SafeGetItemProxy(self.__dict__)))
- return hash_func
- def set_model_fields(
- cls: type[BaseModel],
- bases: tuple[type[Any], ...],
- config_wrapper: ConfigWrapper,
- ns_resolver: NsResolver | None,
- ) -> None:
- """Collect and set `cls.__pydantic_fields__` and `cls.__class_vars__`.
- Args:
- cls: BaseModel or dataclass.
- bases: Parents of the class, generally `cls.__bases__`.
- config_wrapper: The config wrapper instance.
- ns_resolver: Namespace resolver to use when getting model annotations.
- """
- typevars_map = get_model_typevars_map(cls)
- fields, class_vars = collect_model_fields(cls, bases, config_wrapper, ns_resolver, typevars_map=typevars_map)
- cls.__pydantic_fields__ = fields
- cls.__class_vars__.update(class_vars)
- for k in class_vars:
- # Class vars should not be private attributes
- # We remove them _here_ and not earlier because we rely on inspecting the class to determine its classvars,
- # but private attributes are determined by inspecting the namespace _prior_ to class creation.
- # In the case that a classvar with a leading-'_' is defined via a ForwardRef (e.g., when using
- # `__future__.annotations`), we want to remove the private attribute which was detected _before_ we knew it
- # evaluated to a classvar
- value = cls.__private_attributes__.pop(k, None)
- if value is not None and value.default is not PydanticUndefined:
- setattr(cls, k, value.default)
- def complete_model_class(
- cls: type[BaseModel],
- cls_name: str,
- config_wrapper: ConfigWrapper,
- *,
- raise_errors: bool = True,
- ns_resolver: NsResolver | None = None,
- create_model_module: str | None = None,
- ) -> bool:
- """Finish building a model class.
- This logic must be called after class has been created since validation functions must be bound
- and `get_type_hints` requires a class object.
- Args:
- cls: BaseModel or dataclass.
- cls_name: The model or dataclass name.
- config_wrapper: The config wrapper instance.
- raise_errors: Whether to raise errors.
- ns_resolver: The namespace resolver instance to use during schema building.
- create_model_module: The module of the class to be created, if created by `create_model`.
- Returns:
- `True` if the model is successfully completed, else `False`.
- Raises:
- PydanticUndefinedAnnotation: If `PydanticUndefinedAnnotation` occurs in`__get_pydantic_core_schema__`
- and `raise_errors=True`.
- """
- if config_wrapper.defer_build:
- set_model_mocks(cls, cls_name)
- return False
- typevars_map = get_model_typevars_map(cls)
- gen_schema = GenerateSchema(
- config_wrapper,
- ns_resolver,
- typevars_map,
- )
- handler = CallbackGetCoreSchemaHandler(
- partial(gen_schema.generate_schema, from_dunder_get_core_schema=False),
- gen_schema,
- ref_mode='unpack',
- )
- try:
- schema = cls.__get_pydantic_core_schema__(cls, handler)
- except PydanticUndefinedAnnotation as e:
- if raise_errors:
- raise
- set_model_mocks(cls, cls_name, f'`{e.name}`')
- return False
- core_config = config_wrapper.core_config(title=cls.__name__)
- try:
- schema = gen_schema.clean_schema(schema)
- except gen_schema.CollectedInvalid:
- set_model_mocks(cls, cls_name)
- return False
- # debug(schema)
- cls.__pydantic_core_schema__ = schema
- cls.__pydantic_validator__ = create_schema_validator(
- schema,
- cls,
- create_model_module or cls.__module__,
- cls.__qualname__,
- 'create_model' if create_model_module else 'BaseModel',
- core_config,
- config_wrapper.plugin_settings,
- )
- cls.__pydantic_serializer__ = SchemaSerializer(schema, core_config)
- cls.__pydantic_complete__ = True
- # set __signature__ attr only for model class, but not for its instances
- # (because instances can define `__call__`, and `inspect.signature` shouldn't
- # use the `__signature__` attribute and instead generate from `__call__`).
- cls.__signature__ = LazyClassAttribute(
- '__signature__',
- partial(
- generate_pydantic_signature,
- init=cls.__init__,
- fields=cls.__pydantic_fields__,
- populate_by_name=config_wrapper.populate_by_name,
- extra=config_wrapper.extra,
- ),
- )
- return True
- def set_deprecated_descriptors(cls: type[BaseModel]) -> None:
- """Set data descriptors on the class for deprecated fields."""
- for field, field_info in cls.__pydantic_fields__.items():
- if (msg := field_info.deprecation_message) is not None:
- desc = _DeprecatedFieldDescriptor(msg)
- desc.__set_name__(cls, field)
- setattr(cls, field, desc)
- for field, computed_field_info in cls.__pydantic_computed_fields__.items():
- if (
- (msg := computed_field_info.deprecation_message) is not None
- # Avoid having two warnings emitted:
- and not hasattr(unwrap_wrapped_function(computed_field_info.wrapped_property), '__deprecated__')
- ):
- desc = _DeprecatedFieldDescriptor(msg, computed_field_info.wrapped_property)
- desc.__set_name__(cls, field)
- setattr(cls, field, desc)
- class _DeprecatedFieldDescriptor:
- """Read-only data descriptor used to emit a runtime deprecation warning before accessing a deprecated field.
- Attributes:
- msg: The deprecation message to be emitted.
- wrapped_property: The property instance if the deprecated field is a computed field, or `None`.
- field_name: The name of the field being deprecated.
- """
- field_name: str
- def __init__(self, msg: str, wrapped_property: property | None = None) -> None:
- self.msg = msg
- self.wrapped_property = wrapped_property
- def __set_name__(self, cls: type[BaseModel], name: str) -> None:
- self.field_name = name
- def __get__(self, obj: BaseModel | None, obj_type: type[BaseModel] | None = None) -> Any:
- if obj is None:
- if self.wrapped_property is not None:
- return self.wrapped_property.__get__(None, obj_type)
- raise AttributeError(self.field_name)
- warnings.warn(self.msg, builtins.DeprecationWarning, stacklevel=2)
- if self.wrapped_property is not None:
- return self.wrapped_property.__get__(obj, obj_type)
- return obj.__dict__[self.field_name]
- # Defined to make it a data descriptor and take precedence over the instance's dictionary.
- # Note that it will not be called when setting a value on a model instance
- # as `BaseModel.__setattr__` is defined and takes priority.
- def __set__(self, obj: Any, value: Any) -> NoReturn:
- raise AttributeError(self.field_name)
- class _PydanticWeakRef:
- """Wrapper for `weakref.ref` that enables `pickle` serialization.
- Cloudpickle fails to serialize `weakref.ref` objects due to an arcane error related
- to abstract base classes (`abc.ABC`). This class works around the issue by wrapping
- `weakref.ref` instead of subclassing it.
- See https://github.com/pydantic/pydantic/issues/6763 for context.
- Semantics:
- - If not pickled, behaves the same as a `weakref.ref`.
- - If pickled along with the referenced object, the same `weakref.ref` behavior
- will be maintained between them after unpickling.
- - If pickled without the referenced object, after unpickling the underlying
- reference will be cleared (`__call__` will always return `None`).
- """
- def __init__(self, obj: Any):
- if obj is None:
- # The object will be `None` upon deserialization if the serialized weakref
- # had lost its underlying object.
- self._wr = None
- else:
- self._wr = weakref.ref(obj)
- def __call__(self) -> Any:
- if self._wr is None:
- return None
- else:
- return self._wr()
- def __reduce__(self) -> tuple[Callable, tuple[weakref.ReferenceType | None]]:
- return _PydanticWeakRef, (self(),)
- def build_lenient_weakvaluedict(d: dict[str, Any] | None) -> dict[str, Any] | None:
- """Takes an input dictionary, and produces a new value that (invertibly) replaces the values with weakrefs.
- We can't just use a WeakValueDictionary because many types (including int, str, etc.) can't be stored as values
- in a WeakValueDictionary.
- The `unpack_lenient_weakvaluedict` function can be used to reverse this operation.
- """
- if d is None:
- return None
- result = {}
- for k, v in d.items():
- try:
- proxy = _PydanticWeakRef(v)
- except TypeError:
- proxy = v
- result[k] = proxy
- return result
- def unpack_lenient_weakvaluedict(d: dict[str, Any] | None) -> dict[str, Any] | None:
- """Inverts the transform performed by `build_lenient_weakvaluedict`."""
- if d is None:
- return None
- result = {}
- for k, v in d.items():
- if isinstance(v, _PydanticWeakRef):
- v = v()
- if v is not None:
- result[k] = v
- else:
- result[k] = v
- return result
- @lru_cache(maxsize=None)
- def default_ignored_types() -> tuple[type[Any], ...]:
- from ..fields import ComputedFieldInfo
- ignored_types = [
- FunctionType,
- property,
- classmethod,
- staticmethod,
- PydanticDescriptorProxy,
- ComputedFieldInfo,
- TypeAliasType, # from `typing_extensions`
- ]
- if sys.version_info >= (3, 12):
- ignored_types.append(typing.TypeAliasType)
- return tuple(ignored_types)
|