Загрузить файлы в «venv/Lib/site-packages/pydantic/_internal»

This commit is contained in:
2026-07-02 20:00:40 +00:00
parent 65ca76214e
commit 32e458e67f
5 changed files with 972 additions and 0 deletions

Binary file not shown.

View File

@@ -0,0 +1,386 @@
from __future__ import annotations as _annotations
import warnings
from contextlib import contextmanager
from re import Pattern
from typing import (
TYPE_CHECKING,
Any,
Callable,
Literal,
cast,
)
from pydantic_core import core_schema
from typing_extensions import Self
from ..aliases import AliasGenerator
from ..config import ConfigDict, ExtraValues, JsonDict, JsonEncoder, JsonSchemaExtraCallable
from ..errors import PydanticUserError
from ..warnings import PydanticDeprecatedSince20, PydanticDeprecatedSince210
if TYPE_CHECKING:
from .._internal._schema_generation_shared import GenerateSchema
from ..fields import ComputedFieldInfo, FieldInfo
DEPRECATION_MESSAGE = 'Support for class-based `config` is deprecated, use ConfigDict instead.'
class ConfigWrapper:
"""Internal wrapper for Config which exposes ConfigDict items as attributes."""
__slots__ = ('config_dict',)
config_dict: ConfigDict
# all annotations are copied directly from ConfigDict, and should be kept up to date, a test will fail if they
# stop matching
title: str | None
str_to_lower: bool
str_to_upper: bool
str_strip_whitespace: bool
str_min_length: int
str_max_length: int | None
extra: ExtraValues | None
frozen: bool
populate_by_name: bool
use_enum_values: bool
validate_assignment: bool
arbitrary_types_allowed: bool
from_attributes: bool
# whether to use the actual key provided in the data (e.g. alias or first alias for "field required" errors) instead of field_names
# to construct error `loc`s, default `True`
loc_by_alias: bool
alias_generator: Callable[[str], str] | AliasGenerator | None
model_title_generator: Callable[[type], str] | None
field_title_generator: Callable[[str, FieldInfo | ComputedFieldInfo], str] | None
ignored_types: tuple[type, ...]
allow_inf_nan: bool
json_schema_extra: JsonDict | JsonSchemaExtraCallable | None
json_encoders: dict[type[object], JsonEncoder] | None
# new in V2
strict: bool
# whether instances of models and dataclasses (including subclass instances) should re-validate, default 'never'
revalidate_instances: Literal['always', 'never', 'subclass-instances']
ser_json_timedelta: Literal['iso8601', 'float']
ser_json_temporal: Literal['iso8601', 'seconds', 'milliseconds']
val_temporal_unit: Literal['seconds', 'milliseconds', 'infer']
ser_json_bytes: Literal['utf8', 'base64', 'hex']
val_json_bytes: Literal['utf8', 'base64', 'hex']
ser_json_inf_nan: Literal['null', 'constants', 'strings']
# whether to validate default values during validation, default False
validate_default: bool
validate_return: bool
protected_namespaces: tuple[str | Pattern[str], ...]
hide_input_in_errors: bool
defer_build: bool
plugin_settings: dict[str, object] | None
schema_generator: type[GenerateSchema] | None
json_schema_serialization_defaults_required: bool
json_schema_mode_override: Literal['validation', 'serialization', None]
coerce_numbers_to_str: bool
regex_engine: Literal['rust-regex', 'python-re']
validation_error_cause: bool
use_attribute_docstrings: bool
cache_strings: bool | Literal['all', 'keys', 'none']
validate_by_alias: bool
validate_by_name: bool
serialize_by_alias: bool
url_preserve_empty_path: bool
polymorphic_serialization: bool
def __init__(self, config: ConfigDict | dict[str, Any] | type[Any] | None, *, check: bool = True):
if check:
self.config_dict = prepare_config(config)
else:
self.config_dict = cast(ConfigDict, config)
@classmethod
def for_model(
cls,
bases: tuple[type[Any], ...],
namespace: dict[str, Any],
raw_annotations: dict[str, Any],
kwargs: dict[str, Any],
) -> Self:
"""Build a new `ConfigWrapper` instance for a `BaseModel`.
The config wrapper built based on (in descending order of priority):
- options from `kwargs`
- options from the `namespace`
- options from the base classes (`bases`)
Args:
bases: A tuple of base classes.
namespace: The namespace of the class being created.
raw_annotations: The (non-evaluated) annotations of the model.
kwargs: The kwargs passed to the class being created.
Returns:
A `ConfigWrapper` instance for `BaseModel`.
"""
config_new = ConfigDict()
for base in bases:
config = getattr(base, 'model_config', None)
if config:
config_new.update(config.copy())
config_class_from_namespace = namespace.get('Config')
config_dict_from_namespace = namespace.get('model_config')
if raw_annotations.get('model_config') and config_dict_from_namespace is None:
raise PydanticUserError(
'`model_config` cannot be used as a model field name. Use `model_config` for model configuration.',
code='model-config-invalid-field-name',
)
if config_class_from_namespace and config_dict_from_namespace:
raise PydanticUserError('"Config" and "model_config" cannot be used together', code='config-both')
config_from_namespace = config_dict_from_namespace or prepare_config(config_class_from_namespace)
config_new.update(config_from_namespace)
for k in list(kwargs.keys()):
if k in config_keys:
config_new[k] = kwargs.pop(k)
return cls(config_new)
# we don't show `__getattr__` to type checkers so missing attributes cause errors
if not TYPE_CHECKING: # pragma: no branch
def __getattr__(self, name: str) -> Any:
try:
return self.config_dict[name]
except KeyError:
try:
return config_defaults[name]
except KeyError:
raise AttributeError(f'Config has no attribute {name!r}') from None
def core_config(self, title: str | None) -> core_schema.CoreConfig:
"""Create a pydantic-core config.
We don't use getattr here since we don't want to populate with defaults.
Args:
title: The title to use if not set in config.
Returns:
A `CoreConfig` object created from config.
"""
config = self.config_dict
if config.get('schema_generator') is not None:
warnings.warn(
'The `schema_generator` setting has been deprecated since v2.10. This setting no longer has any effect.',
PydanticDeprecatedSince210,
stacklevel=2,
)
if (populate_by_name := config.get('populate_by_name')) is not None:
# We include this patch for backwards compatibility purposes, but this config setting will be deprecated in v3.0, and likely removed in v4.0.
# Thus, the above warning and this patch can be removed then as well.
if config.get('validate_by_name') is None:
config['validate_by_alias'] = True
config['validate_by_name'] = populate_by_name
# We dynamically patch validate_by_name to be True if validate_by_alias is set to False
# and validate_by_name is not explicitly set.
if config.get('validate_by_alias') is False and config.get('validate_by_name') is None:
config['validate_by_name'] = True
if (not config.get('validate_by_alias', True)) and (not config.get('validate_by_name', False)):
raise PydanticUserError(
'At least one of `validate_by_alias` or `validate_by_name` must be set to True.',
code='validate-by-alias-and-name-false',
)
return core_schema.CoreConfig(
**{ # pyright: ignore[reportArgumentType]
k: v
for k, v in (
('title', config.get('title') or title or None),
('extra_fields_behavior', config.get('extra')),
('allow_inf_nan', config.get('allow_inf_nan')),
('str_strip_whitespace', config.get('str_strip_whitespace')),
('str_to_lower', config.get('str_to_lower')),
('str_to_upper', config.get('str_to_upper')),
('strict', config.get('strict')),
('ser_json_timedelta', config.get('ser_json_timedelta')),
('ser_json_temporal', config.get('ser_json_temporal')),
('val_temporal_unit', config.get('val_temporal_unit')),
('ser_json_bytes', config.get('ser_json_bytes')),
('val_json_bytes', config.get('val_json_bytes')),
('ser_json_inf_nan', config.get('ser_json_inf_nan')),
('from_attributes', config.get('from_attributes')),
('loc_by_alias', config.get('loc_by_alias')),
('revalidate_instances', config.get('revalidate_instances')),
('validate_default', config.get('validate_default')),
('str_max_length', config.get('str_max_length')),
('str_min_length', config.get('str_min_length')),
('hide_input_in_errors', config.get('hide_input_in_errors')),
('coerce_numbers_to_str', config.get('coerce_numbers_to_str')),
('regex_engine', config.get('regex_engine')),
('validation_error_cause', config.get('validation_error_cause')),
('cache_strings', config.get('cache_strings')),
('validate_by_alias', config.get('validate_by_alias')),
('validate_by_name', config.get('validate_by_name')),
('serialize_by_alias', config.get('serialize_by_alias')),
('url_preserve_empty_path', config.get('url_preserve_empty_path')),
('polymorphic_serialization', config.get('polymorphic_serialization')),
)
if v is not None
}
)
def __repr__(self):
c = ', '.join(f'{k}={v!r}' for k, v in self.config_dict.items())
return f'ConfigWrapper({c})'
class ConfigWrapperStack:
"""A stack of `ConfigWrapper` instances."""
def __init__(self, config_wrapper: ConfigWrapper):
self._config_wrapper_stack: list[ConfigWrapper] = [config_wrapper]
@property
def tail(self) -> ConfigWrapper:
return self._config_wrapper_stack[-1]
@contextmanager
def push(self, config_wrapper: ConfigWrapper | ConfigDict | None):
if config_wrapper is None:
yield
return
if not isinstance(config_wrapper, ConfigWrapper):
config_wrapper = ConfigWrapper(config_wrapper, check=False)
self._config_wrapper_stack.append(config_wrapper)
try:
yield
finally:
self._config_wrapper_stack.pop()
config_defaults = ConfigDict(
title=None,
str_to_lower=False,
str_to_upper=False,
str_strip_whitespace=False,
str_min_length=0,
str_max_length=None,
# let the model / dataclass decide how to handle it
extra=None,
frozen=False,
populate_by_name=False,
use_enum_values=False,
validate_assignment=False,
arbitrary_types_allowed=False,
from_attributes=False,
loc_by_alias=True,
alias_generator=None,
model_title_generator=None,
field_title_generator=None,
ignored_types=(),
allow_inf_nan=True,
json_schema_extra=None,
strict=False,
revalidate_instances='never',
ser_json_timedelta='iso8601',
ser_json_temporal='iso8601',
val_temporal_unit='infer',
ser_json_bytes='utf8',
val_json_bytes='utf8',
ser_json_inf_nan='null',
validate_default=False,
validate_return=False,
protected_namespaces=('model_validate', 'model_dump'),
hide_input_in_errors=False,
json_encoders=None,
defer_build=False,
schema_generator=None,
plugin_settings=None,
json_schema_serialization_defaults_required=False,
json_schema_mode_override=None,
coerce_numbers_to_str=False,
regex_engine='rust-regex',
validation_error_cause=False,
use_attribute_docstrings=False,
cache_strings=True,
validate_by_alias=True,
validate_by_name=False,
serialize_by_alias=False,
url_preserve_empty_path=False,
polymorphic_serialization=False,
)
def prepare_config(config: ConfigDict | dict[str, Any] | type[Any] | None) -> ConfigDict:
"""Create a `ConfigDict` instance from an existing dict, a class (e.g. old class-based config) or None.
Args:
config: The input config.
Returns:
A ConfigDict object created from config.
"""
if config is None:
return ConfigDict()
if not isinstance(config, dict):
warnings.warn(DEPRECATION_MESSAGE, PydanticDeprecatedSince20, stacklevel=4)
config = {k: getattr(config, k) for k in dir(config) if not k.startswith('__')}
config_dict = cast(ConfigDict, config)
check_deprecated(config_dict)
return config_dict
config_keys = set(ConfigDict.__annotations__.keys())
V2_REMOVED_KEYS = {
'allow_mutation',
'error_msg_templates',
'fields',
'getter_dict',
'smart_union',
'underscore_attrs_are_private',
'json_loads',
'json_dumps',
'copy_on_model_validation',
'post_init_call',
}
V2_RENAMED_KEYS = {
'allow_population_by_field_name': 'validate_by_name',
'anystr_lower': 'str_to_lower',
'anystr_strip_whitespace': 'str_strip_whitespace',
'anystr_upper': 'str_to_upper',
'keep_untouched': 'ignored_types',
'max_anystr_length': 'str_max_length',
'min_anystr_length': 'str_min_length',
'orm_mode': 'from_attributes',
'schema_extra': 'json_schema_extra',
'validate_all': 'validate_default',
}
def check_deprecated(config_dict: ConfigDict) -> None:
"""Check for deprecated config keys and warn the user.
Args:
config_dict: The input config.
"""
deprecated_removed_keys = V2_REMOVED_KEYS & config_dict.keys()
deprecated_renamed_keys = V2_RENAMED_KEYS.keys() & config_dict.keys()
if deprecated_removed_keys or deprecated_renamed_keys:
renamings = {k: V2_RENAMED_KEYS[k] for k in sorted(deprecated_renamed_keys)}
renamed_bullets = [f'* {k!r} has been renamed to {v!r}' for k, v in renamings.items()]
removed_bullets = [f'* {k!r} has been removed' for k in sorted(deprecated_removed_keys)]
message = '\n'.join(['Valid config keys have changed in V2:'] + renamed_bullets + removed_bullets)
warnings.warn(message, UserWarning)

View File

@@ -0,0 +1,97 @@
from __future__ import annotations as _annotations
from typing import TYPE_CHECKING, Any, TypedDict, cast
from warnings import warn
if TYPE_CHECKING:
from ..config import JsonDict, JsonSchemaExtraCallable
from ._schema_generation_shared import (
GetJsonSchemaFunction,
)
class CoreMetadata(TypedDict, total=False):
"""A `TypedDict` for holding the metadata dict of the schema.
Attributes:
pydantic_js_functions: List of JSON schema functions that resolve refs during application.
pydantic_js_annotation_functions: List of JSON schema functions that don't resolve refs during application.
pydantic_js_prefer_positional_arguments: Whether JSON schema generator will
prefer positional over keyword arguments for an 'arguments' schema.
custom validation function. Only applies to before, plain, and wrap validators.
pydantic_js_updates: key / value pair updates to apply to the JSON schema for a type.
pydantic_js_extra: WIP, either key/value pair updates to apply to the JSON schema, or a custom callable.
pydantic_internal_union_tag_key: Used internally by the `Tag` metadata to specify the tag used for a discriminated union.
pydantic_internal_union_discriminator: Used internally to specify the discriminator value for a discriminated union
when the discriminator was applied to a `'definition-ref'` schema, and that reference was missing at the time
of the annotation application.
TODO: Perhaps we should move this structure to pydantic-core. At the moment, though,
it's easier to iterate on if we leave it in pydantic until we feel there is a semi-stable API.
TODO: It's unfortunate how functionally oriented JSON schema generation is, especially that which occurs during
the core schema generation process. It's inevitable that we need to store some json schema related information
on core schemas, given that we generate JSON schemas directly from core schemas. That being said, debugging related
issues is quite difficult when JSON schema information is disguised via dynamically defined functions.
"""
pydantic_js_functions: list[GetJsonSchemaFunction]
pydantic_js_annotation_functions: list[GetJsonSchemaFunction]
pydantic_js_prefer_positional_arguments: bool
pydantic_js_updates: JsonDict
pydantic_js_extra: JsonDict | JsonSchemaExtraCallable
pydantic_internal_union_tag_key: str
pydantic_internal_union_discriminator: str
def update_core_metadata(
core_metadata: Any,
/,
*,
pydantic_js_functions: list[GetJsonSchemaFunction] | None = None,
pydantic_js_annotation_functions: list[GetJsonSchemaFunction] | None = None,
pydantic_js_updates: JsonDict | None = None,
pydantic_js_extra: JsonDict | JsonSchemaExtraCallable | None = None,
) -> None:
from ..json_schema import PydanticJsonSchemaWarning
"""Update CoreMetadata instance in place. When we make modifications in this function, they
take effect on the `core_metadata` reference passed in as the first (and only) positional argument.
First, cast to `CoreMetadata`, then finish with a cast to `dict[str, Any]` for core schema compatibility.
We do this here, instead of before / after each call to this function so that this typing hack
can be easily removed if/when we move `CoreMetadata` to `pydantic-core`.
For parameter descriptions, see `CoreMetadata` above.
"""
core_metadata = cast(CoreMetadata, core_metadata)
if pydantic_js_functions:
core_metadata.setdefault('pydantic_js_functions', []).extend(pydantic_js_functions)
if pydantic_js_annotation_functions:
core_metadata.setdefault('pydantic_js_annotation_functions', []).extend(pydantic_js_annotation_functions)
if pydantic_js_updates:
if (existing_updates := core_metadata.get('pydantic_js_updates')) is not None:
core_metadata['pydantic_js_updates'] = {**existing_updates, **pydantic_js_updates}
else:
core_metadata['pydantic_js_updates'] = pydantic_js_updates
if pydantic_js_extra is not None:
existing_pydantic_js_extra = core_metadata.get('pydantic_js_extra')
if existing_pydantic_js_extra is None:
core_metadata['pydantic_js_extra'] = pydantic_js_extra
if isinstance(existing_pydantic_js_extra, dict):
if isinstance(pydantic_js_extra, dict):
core_metadata['pydantic_js_extra'] = {**existing_pydantic_js_extra, **pydantic_js_extra}
if callable(pydantic_js_extra):
warn(
'Composing `dict` and `callable` type `json_schema_extra` is not supported.'
'The `callable` type is being ignored.'
"If you'd like support for this behavior, please open an issue on pydantic.",
PydanticJsonSchemaWarning,
)
if callable(existing_pydantic_js_extra):
# if ever there's a case of a callable, we'll just keep the last json schema extra spec
core_metadata['pydantic_js_extra'] = pydantic_js_extra

View File

@@ -0,0 +1,174 @@
from __future__ import annotations
import inspect
from collections.abc import Mapping, Sequence
from typing import TYPE_CHECKING, Any, Union
from pydantic_core import CoreSchema, core_schema
from typing_extensions import TypeGuard, get_args, get_origin
from typing_inspection import typing_objects
from . import _repr
from ._typing_extra import is_generic_alias
if TYPE_CHECKING:
from rich.console import Console
AnyFunctionSchema = Union[
core_schema.AfterValidatorFunctionSchema,
core_schema.BeforeValidatorFunctionSchema,
core_schema.WrapValidatorFunctionSchema,
core_schema.PlainValidatorFunctionSchema,
]
FunctionSchemaWithInnerSchema = Union[
core_schema.AfterValidatorFunctionSchema,
core_schema.BeforeValidatorFunctionSchema,
core_schema.WrapValidatorFunctionSchema,
]
CoreSchemaField = Union[
core_schema.ModelField, core_schema.DataclassField, core_schema.TypedDictField, core_schema.ComputedField
]
CoreSchemaOrField = Union[core_schema.CoreSchema, CoreSchemaField]
_CORE_SCHEMA_FIELD_TYPES = {'typed-dict-field', 'dataclass-field', 'model-field', 'computed-field'}
_FUNCTION_WITH_INNER_SCHEMA_TYPES = {'function-before', 'function-after', 'function-wrap'}
_LIST_LIKE_SCHEMA_WITH_ITEMS_TYPES = {'list', 'set', 'frozenset'}
def is_core_schema(
schema: CoreSchemaOrField,
) -> TypeGuard[CoreSchema]:
return schema['type'] not in _CORE_SCHEMA_FIELD_TYPES
def is_core_schema_field(
schema: CoreSchemaOrField,
) -> TypeGuard[CoreSchemaField]:
return schema['type'] in _CORE_SCHEMA_FIELD_TYPES
def is_function_with_inner_schema(
schema: CoreSchemaOrField,
) -> TypeGuard[FunctionSchemaWithInnerSchema]:
return schema['type'] in _FUNCTION_WITH_INNER_SCHEMA_TYPES
def is_list_like_schema_with_items_schema(
schema: CoreSchema,
) -> TypeGuard[core_schema.ListSchema | core_schema.SetSchema | core_schema.FrozenSetSchema]:
return schema['type'] in _LIST_LIKE_SCHEMA_WITH_ITEMS_TYPES
def get_type_ref(type_: Any, args_override: tuple[type[Any], ...] | None = None) -> str:
"""Produces the ref to be used for this type by pydantic_core's core schemas.
This `args_override` argument was added for the purpose of creating valid recursive references
when creating generic models without needing to create a concrete class.
"""
origin = get_origin(type_) or type_
args = get_args(type_) if is_generic_alias(type_) else (args_override or ())
generic_metadata = getattr(type_, '__pydantic_generic_metadata__', None)
if generic_metadata:
origin = generic_metadata['origin'] or origin
args = generic_metadata['args'] or args
module_name = getattr(origin, '__module__', '<No __module__>')
if typing_objects.is_typealiastype(origin):
type_ref = f'{module_name}.{origin.__name__}:{id(origin)}'
else:
try:
qualname = getattr(origin, '__qualname__', f'<No __qualname__: {origin}>')
except Exception:
qualname = getattr(origin, '__qualname__', '<No __qualname__>')
type_ref = f'{module_name}.{qualname}:{id(origin)}'
arg_refs: list[str] = []
for arg in args:
if isinstance(arg, str):
# Handle string literals as a special case; we may be able to remove this special handling if we
# wrap them in a ForwardRef at some point.
arg_ref = f'{arg}:str-{id(arg)}'
else:
arg_ref = f'{_repr.display_as_type(arg)}:{id(arg)}'
arg_refs.append(arg_ref)
if arg_refs:
type_ref = f'{type_ref}[{",".join(arg_refs)}]'
return type_ref
def get_ref(s: core_schema.CoreSchema) -> None | str:
"""Get the ref from the schema if it has one.
This exists just for type checking to work correctly.
"""
return s.get('ref', None)
def _clean_schema_for_pretty_print(obj: Any, strip_metadata: bool = True) -> Any: # pragma: no cover
"""A utility function to remove irrelevant information from a core schema."""
if isinstance(obj, Mapping):
new_dct = {}
for k, v in obj.items():
if k == 'metadata' and strip_metadata:
new_metadata = {}
for meta_k, meta_v in v.items():
if meta_k in ('pydantic_js_functions', 'pydantic_js_annotation_functions'):
new_metadata['js_metadata'] = '<stripped>'
else:
new_metadata[meta_k] = _clean_schema_for_pretty_print(meta_v, strip_metadata=strip_metadata)
if list(new_metadata.keys()) == ['js_metadata']:
new_metadata = {'<stripped>'}
new_dct[k] = new_metadata
# Remove some defaults:
elif k in ('custom_init', 'root_model') and not v:
continue
else:
new_dct[k] = _clean_schema_for_pretty_print(v, strip_metadata=strip_metadata)
return new_dct
elif isinstance(obj, Sequence) and not isinstance(obj, str):
return [_clean_schema_for_pretty_print(v, strip_metadata=strip_metadata) for v in obj]
else:
return obj
def pretty_print_core_schema(
val: Any,
*,
console: Console | None = None,
max_depth: int | None = None,
strip_metadata: bool = True,
) -> None: # pragma: no cover
"""Pretty-print a core schema using the `rich` library.
Args:
val: The core schema to print, or a Pydantic model/dataclass/type adapter
(in which case the cached core schema is fetched and printed).
console: A rich console to use when printing. Defaults to the global rich console instance.
max_depth: The number of nesting levels which may be printed.
strip_metadata: Whether to strip metadata in the output. If `True` any known core metadata
attributes will be stripped (but custom attributes are kept). Defaults to `True`.
"""
# lazy import:
from rich.pretty import pprint
# circ. imports:
from pydantic import BaseModel, TypeAdapter
from pydantic.dataclasses import is_pydantic_dataclass
if (inspect.isclass(val) and issubclass(val, BaseModel)) or is_pydantic_dataclass(val):
val = val.__pydantic_core_schema__
if isinstance(val, TypeAdapter):
val = val.core_schema
cleaned_schema = _clean_schema_for_pretty_print(val, strip_metadata=strip_metadata)
pprint(cleaned_schema, console=console, max_depth=max_depth)
pps = pretty_print_core_schema

View File

@@ -0,0 +1,315 @@
"""Private logic for creating pydantic dataclasses."""
from __future__ import annotations as _annotations
import copy
import dataclasses
import sys
import warnings
from collections.abc import Generator
from contextlib import contextmanager
from functools import partial
from typing import TYPE_CHECKING, Any, ClassVar, Protocol, cast
from pydantic_core import (
ArgsKwargs,
SchemaSerializer,
SchemaValidator,
core_schema,
)
from typing_extensions import TypeAlias, TypeIs
from ..errors import PydanticUndefinedAnnotation
from ..fields import FieldInfo
from ..plugin._schema_validator import PluggableSchemaValidator, create_schema_validator
from ..warnings import PydanticDeprecatedSince20
from . import _config, _decorators
from ._fields import collect_dataclass_fields
from ._generate_schema import GenerateSchema, InvalidSchemaError
from ._generics import get_standard_typevars_map
from ._mock_val_ser import set_dataclass_mocks
from ._namespace_utils import NsResolver
from ._signature import generate_pydantic_signature
from ._utils import LazyClassAttribute
if TYPE_CHECKING:
from _typeshed import DataclassInstance as StandardDataclass
from ..config import ConfigDict
class PydanticDataclass(StandardDataclass, Protocol):
"""A protocol containing attributes only available once a class has been decorated as a Pydantic dataclass.
Attributes:
__pydantic_config__: Pydantic-specific configuration settings for the dataclass.
__pydantic_complete__: Whether dataclass building is completed, or if there are still undefined fields.
__pydantic_core_schema__: The pydantic-core schema used to build the SchemaValidator and SchemaSerializer.
__pydantic_decorators__: Metadata containing the decorators defined on the dataclass.
__pydantic_fields__: Metadata about the fields defined on the dataclass.
__pydantic_serializer__: The pydantic-core SchemaSerializer used to dump instances of the dataclass.
__pydantic_validator__: The pydantic-core SchemaValidator used to validate instances of the dataclass.
"""
__pydantic_config__: ClassVar[ConfigDict]
__pydantic_complete__: ClassVar[bool]
__pydantic_core_schema__: ClassVar[core_schema.CoreSchema]
__pydantic_decorators__: ClassVar[_decorators.DecoratorInfos]
__pydantic_fields__: ClassVar[dict[str, FieldInfo]]
__pydantic_serializer__: ClassVar[SchemaSerializer]
__pydantic_validator__: ClassVar[SchemaValidator | PluggableSchemaValidator]
@classmethod
def __pydantic_fields_complete__(cls) -> bool: ...
def set_dataclass_fields(
cls: type[StandardDataclass],
config_wrapper: _config.ConfigWrapper,
ns_resolver: NsResolver | None = None,
) -> None:
"""Collect and set `cls.__pydantic_fields__`.
Args:
cls: The class.
config_wrapper: The config wrapper instance.
ns_resolver: Namespace resolver to use when getting dataclass annotations.
"""
typevars_map = get_standard_typevars_map(cls)
fields = collect_dataclass_fields(
cls, ns_resolver=ns_resolver, typevars_map=typevars_map, config_wrapper=config_wrapper
)
cls.__pydantic_fields__ = fields # type: ignore
def complete_dataclass(
cls: type[Any],
config_wrapper: _config.ConfigWrapper,
*,
raise_errors: bool = True,
ns_resolver: NsResolver | None = None,
_force_build: bool = False,
) -> bool:
"""Finish building a pydantic dataclass.
This logic is called on a class which has already been wrapped in `dataclasses.dataclass()`.
This is somewhat analogous to `pydantic._internal._model_construction.complete_model_class`.
Args:
cls: The class.
config_wrapper: The config wrapper instance.
raise_errors: Whether to raise errors, defaults to `True`.
ns_resolver: The namespace resolver instance to use when collecting dataclass fields
and during schema building.
_force_build: Whether to force building the dataclass, no matter if
[`defer_build`][pydantic.config.ConfigDict.defer_build] is set.
Returns:
`True` if building a pydantic dataclass is successfully completed, `False` otherwise.
Raises:
PydanticUndefinedAnnotation: If `raise_error` is `True` and there is an undefined annotations.
"""
original_init = cls.__init__
# dataclass.__init__ must be defined here so its `__qualname__` can be changed since functions can't be copied,
# and so that the mock validator is used if building was deferred:
def __init__(__dataclass_self__: PydanticDataclass, *args: Any, **kwargs: Any) -> None:
__tracebackhide__ = True
s = __dataclass_self__
s.__pydantic_validator__.validate_python(ArgsKwargs(args, kwargs), self_instance=s)
__init__.__qualname__ = f'{cls.__qualname__}.__init__'
cls.__init__ = __init__ # type: ignore
cls.__pydantic_config__ = config_wrapper.config_dict # type: ignore
set_dataclass_fields(cls, config_wrapper=config_wrapper, ns_resolver=ns_resolver)
if not _force_build and config_wrapper.defer_build:
set_dataclass_mocks(cls)
return False
if hasattr(cls, '__post_init_post_parse__'):
warnings.warn(
'Support for `__post_init_post_parse__` has been dropped, the method will not be called',
PydanticDeprecatedSince20,
)
typevars_map = get_standard_typevars_map(cls)
gen_schema = GenerateSchema(
config_wrapper,
ns_resolver=ns_resolver,
typevars_map=typevars_map,
)
# set __signature__ attr only for the 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,
# It's important that we reference the `original_init` here,
# as it is the one synthesized by the stdlib `dataclass` module:
init=original_init,
fields=cls.__pydantic_fields__, # type: ignore
validate_by_name=config_wrapper.validate_by_name,
extra=config_wrapper.extra,
is_dataclass=True,
),
)
try:
schema = gen_schema.generate_schema(cls)
except PydanticUndefinedAnnotation as e:
if raise_errors:
raise
set_dataclass_mocks(cls, f'`{e.name}`')
return False
core_config = config_wrapper.core_config(title=cls.__name__)
try:
schema = gen_schema.clean_schema(schema)
except InvalidSchemaError:
set_dataclass_mocks(cls)
return False
# We are about to set all the remaining required properties expected for this cast;
# __pydantic_decorators__ and __pydantic_fields__ should already be set
cls = cast('type[PydanticDataclass]', cls)
cls.__pydantic_core_schema__ = schema
cls.__pydantic_validator__ = create_schema_validator(
schema, cls, cls.__module__, cls.__qualname__, 'dataclass', core_config, config_wrapper.plugin_settings
)
cls.__pydantic_serializer__ = SchemaSerializer(schema, core_config)
cls.__pydantic_complete__ = True
return True
def is_stdlib_dataclass(cls: type[Any], /) -> TypeIs[type[StandardDataclass]]:
"""Returns `True` if the class is a stdlib dataclass and *not* a Pydantic dataclass.
Unlike the stdlib `dataclasses.is_dataclass()` function, this does *not* include subclasses
of a dataclass that are themselves not dataclasses.
Args:
cls: The class.
Returns:
`True` if the class is a stdlib dataclass, `False` otherwise.
"""
return '__dataclass_fields__' in cls.__dict__ and not hasattr(cls, '__pydantic_validator__')
def as_dataclass_field(pydantic_field: FieldInfo) -> dataclasses.Field[Any]:
field_args: dict[str, Any] = {'default': pydantic_field}
# Needed because if `doc` is set, the dataclass slots will be a dict (field name -> doc) instead of a tuple:
if sys.version_info >= (3, 14) and pydantic_field.description is not None:
field_args['doc'] = pydantic_field.description
# Needed as the stdlib dataclass module processes kw_only in a specific way during class construction:
if sys.version_info >= (3, 10) and pydantic_field.kw_only is not None:
field_args['kw_only'] = pydantic_field.kw_only
# Needed as the stdlib dataclass modules generates `__repr__()` during class construction:
if pydantic_field.repr is not True:
field_args['repr'] = pydantic_field.repr
return dataclasses.field(**field_args)
DcFields: TypeAlias = dict[str, dataclasses.Field[Any]]
@contextmanager
def patch_base_fields(cls: type[Any]) -> Generator[None]:
"""Temporarily patch the stdlib dataclasses bases of `cls` if the Pydantic `Field()` function is used.
When creating a Pydantic dataclass, it is possible to inherit from stdlib dataclasses, where
the Pydantic `Field()` function is used. To create this Pydantic dataclass, we first apply
the stdlib `@dataclass` decorator on it. During the construction of the stdlib dataclass,
the `kw_only` and `repr` field arguments need to be understood by the stdlib *during* the
dataclass construction. To do so, we temporarily patch the fields dictionary of the affected
bases.
For instance, with the following example:
```python {test="skip" lint="skip"}
import dataclasses as stdlib_dc
import pydantic
import pydantic.dataclasses as pydantic_dc
@stdlib_dc.dataclass
class A:
a: int = pydantic.Field(repr=False)
# Notice that the `repr` attribute of the dataclass field is `True`:
A.__dataclass_fields__['a']
#> dataclass.Field(default=FieldInfo(repr=False), repr=True, ...)
@pydantic_dc.dataclass
class B(A):
b: int = pydantic.Field(repr=False)
```
When passing `B` to the stdlib `@dataclass` decorator, it will look for fields in the parent classes
and reuse them directly. When this context manager is active, `A` will be temporarily patched to be
equivalent to:
```python {test="skip" lint="skip"}
@stdlib_dc.dataclass
class A:
a: int = stdlib_dc.field(default=Field(repr=False), repr=False)
```
!!! note
This is only applied to the bases of `cls`, and not `cls` itself. The reason is that the Pydantic
dataclass decorator "owns" `cls` (in the previous example, `B`). As such, we instead modify the fields
directly (in the previous example, we simply do `setattr(B, 'b', as_dataclass_field(pydantic_field))`).
!!! note
This approach is far from ideal, and can probably be the source of unwanted side effects/race conditions.
The previous implemented approach was mutating the `__annotations__` dict of `cls`, which is no longer a
safe operation in Python 3.14+, and resulted in unexpected behavior with field ordering anyway.
"""
# A list of two-tuples, the first element being a reference to the
# dataclass fields dictionary, the second element being a mapping between
# the field names that were modified, and their original `Field`:
original_fields_list: list[tuple[DcFields, DcFields]] = []
for base in cls.__mro__[1:]:
dc_fields: dict[str, dataclasses.Field[Any]] = base.__dict__.get('__dataclass_fields__', {})
dc_fields_with_pydantic_field_defaults = {
field_name: field
for field_name, field in dc_fields.items()
if isinstance(field.default, FieldInfo)
# Only do the patching if one of the affected attributes is set:
and (field.default.description is not None or field.default.kw_only or field.default.repr is not True)
}
if dc_fields_with_pydantic_field_defaults:
original_fields_list.append((dc_fields, dc_fields_with_pydantic_field_defaults))
for field_name, field in dc_fields_with_pydantic_field_defaults.items():
default = cast(FieldInfo, field.default)
# `dataclasses.Field` isn't documented as working with `copy.copy()`.
# It is a class with `__slots__`, so should work (and we hope for the best):
new_dc_field = copy.copy(field)
# For base fields, no need to set `doc` from `FieldInfo.description`, this is only relevant
# for the class under construction and handled in `as_dataclass_field()`.
if sys.version_info >= (3, 10) and default.kw_only:
new_dc_field.kw_only = True
if default.repr is not True:
new_dc_field.repr = default.repr
dc_fields[field_name] = new_dc_field
try:
yield
finally:
for fields, original_fields in original_fields_list:
for field_name, original_field in original_fields.items():
fields[field_name] = original_field