Coverage for pydantic/_internal/_generate_schema.py: 95.65%
1120 statements
« prev ^ index » next coverage.py v7.5.3, created at 2024-06-21 17:00 +0000
1"""Convert python types to pydantic-core schema."""
3from __future__ import annotations as _annotations 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
5import collections.abc 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
6import dataclasses 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
7import inspect 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
8import re 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
9import sys 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
10import typing 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
11import warnings 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
12from contextlib import ExitStack, contextmanager 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
13from copy import copy, deepcopy 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
14from enum import Enum 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
15from functools import partial 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
16from inspect import Parameter, _ParameterKind, signature 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
17from itertools import chain 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
18from operator import attrgetter 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
19from types import FunctionType, LambdaType, MethodType 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
20from typing import ( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
21 TYPE_CHECKING,
22 Any,
23 Callable,
24 Dict,
25 Final,
26 ForwardRef,
27 Iterable,
28 Iterator,
29 Mapping,
30 Type,
31 TypeVar,
32 Union,
33 cast,
34 overload,
35)
36from warnings import warn 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
38from pydantic_core import CoreSchema, PydanticUndefined, core_schema, to_jsonable_python 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
39from typing_extensions import Annotated, Literal, TypeAliasType, TypedDict, get_args, get_origin, is_typeddict 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
41from ..aliases import AliasGenerator 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
42from ..annotated_handlers import GetCoreSchemaHandler, GetJsonSchemaHandler 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
43from ..config import ConfigDict, JsonDict, JsonEncoder 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
44from ..errors import PydanticSchemaGenerationError, PydanticUndefinedAnnotation, PydanticUserError 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
45from ..json_schema import JsonSchemaValue 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
46from ..version import version_short 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
47from ..warnings import PydanticDeprecatedSince20 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
48from . import _core_utils, _decorators, _discriminated_union, _known_annotated_metadata, _typing_extra 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
49from ._config import ConfigWrapper, ConfigWrapperStack 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
50from ._core_metadata import CoreMetadataHandler, build_metadata_dict 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
51from ._core_utils import ( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
52 CoreSchemaOrField,
53 collect_invalid_schemas,
54 define_expected_missing_refs,
55 get_ref,
56 get_type_ref,
57 is_function_with_inner_schema,
58 is_list_like_schema_with_items_schema,
59 simplify_schema_references,
60 validate_core_schema,
61)
62from ._decorators import ( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
63 Decorator,
64 DecoratorInfos,
65 FieldSerializerDecoratorInfo,
66 FieldValidatorDecoratorInfo,
67 ModelSerializerDecoratorInfo,
68 ModelValidatorDecoratorInfo,
69 RootValidatorDecoratorInfo,
70 ValidatorDecoratorInfo,
71 get_attribute_from_bases,
72 inspect_field_serializer,
73 inspect_model_serializer,
74 inspect_validator,
75)
76from ._docs_extraction import extract_docstrings_from_cls 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
77from ._fields import collect_dataclass_fields, get_type_hints_infer_globalns 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
78from ._forward_ref import PydanticRecursiveRef 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
79from ._generics import get_standard_typevars_map, has_instance_in_type, recursively_defined_type_refs, replace_types 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
80from ._mock_val_ser import MockCoreSchema 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
81from ._schema_generation_shared import CallbackGetCoreSchemaHandler 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
82from ._typing_extra import is_finalvar, is_self_type 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
83from ._utils import lenient_issubclass 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
85if TYPE_CHECKING: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
86 from ..fields import ComputedFieldInfo, FieldInfo
87 from ..main import BaseModel
88 from ..types import Discriminator
89 from ..validators import FieldValidatorModes
90 from ._dataclasses import StandardDataclass
91 from ._schema_generation_shared import GetJsonSchemaFunction
93_SUPPORTS_TYPEDDICT = sys.version_info >= (3, 12) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
94_AnnotatedType = type(Annotated[int, 123]) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
96FieldDecoratorInfo = Union[ValidatorDecoratorInfo, FieldValidatorDecoratorInfo, FieldSerializerDecoratorInfo] 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
97FieldDecoratorInfoType = TypeVar('FieldDecoratorInfoType', bound=FieldDecoratorInfo) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
98AnyFieldDecorator = Union[ 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
99 Decorator[ValidatorDecoratorInfo],
100 Decorator[FieldValidatorDecoratorInfo],
101 Decorator[FieldSerializerDecoratorInfo],
102]
104ModifyCoreSchemaWrapHandler = GetCoreSchemaHandler 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
105GetCoreSchemaFunction = Callable[[Any, ModifyCoreSchemaWrapHandler], core_schema.CoreSchema] 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
108TUPLE_TYPES: list[type] = [tuple, typing.Tuple] 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
109LIST_TYPES: list[type] = [list, typing.List, collections.abc.MutableSequence] 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
110SET_TYPES: list[type] = [set, typing.Set, collections.abc.MutableSet] 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
111FROZEN_SET_TYPES: list[type] = [frozenset, typing.FrozenSet, collections.abc.Set] 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
112DICT_TYPES: list[type] = [dict, typing.Dict, collections.abc.MutableMapping, collections.abc.Mapping] 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
115def check_validator_fields_against_field_name( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
116 info: FieldDecoratorInfo,
117 field: str,
118) -> bool:
119 """Check if field name is in validator fields.
121 Args:
122 info: The field info.
123 field: The field name to check.
125 Returns:
126 `True` if field name is in validator fields, `False` otherwise.
127 """
128 if '*' in info.fields: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
129 return True 1eiacnqryszkolmfgtuABCDhjbdpvwExF
130 for v_field_name in info.fields: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
131 if v_field_name == field: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
132 return True 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
133 return False 1eiacnqryszkolmfgtuABCDhjbdpvwExF
136def check_decorator_fields_exist(decorators: Iterable[AnyFieldDecorator], fields: Iterable[str]) -> None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
137 """Check if the defined fields in decorators exist in `fields` param.
139 It ignores the check for a decorator if the decorator has `*` as field or `check_fields=False`.
141 Args:
142 decorators: An iterable of decorators.
143 fields: An iterable of fields name.
145 Raises:
146 PydanticUserError: If one of the field names does not exist in `fields` param.
147 """
148 fields = set(fields) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
149 for dec in decorators: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
150 if '*' in dec.info.fields: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
151 continue 1eiacnqryszkolmfgtuABCDhjbdpvwExF
152 if dec.info.check_fields is False: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
153 continue 1eiacnqryszkolmfgtuABCDhjbdpvwExF
154 for field in dec.info.fields: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
155 if field not in fields: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
156 raise PydanticUserError( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
157 f'Decorators defined with incorrect fields: {dec.cls_ref}.{dec.cls_var_name}'
158 " (use check_fields=False if you're inheriting from the model and intended this)",
159 code='decorator-missing-field',
160 )
163def filter_field_decorator_info_by_field( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
164 validator_functions: Iterable[Decorator[FieldDecoratorInfoType]], field: str
165) -> list[Decorator[FieldDecoratorInfoType]]:
166 return [dec for dec in validator_functions if check_validator_fields_against_field_name(dec.info, field)] 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
169def apply_each_item_validators( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
170 schema: core_schema.CoreSchema,
171 each_item_validators: list[Decorator[ValidatorDecoratorInfo]],
172 field_name: str | None,
173) -> core_schema.CoreSchema:
174 # This V1 compatibility shim should eventually be removed
176 # push down any `each_item=True` validators
177 # note that this won't work for any Annotated types that get wrapped by a function validator
178 # but that's okay because that didn't exist in V1
179 if schema['type'] == 'nullable': 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
180 schema['schema'] = apply_each_item_validators(schema['schema'], each_item_validators, field_name) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
181 return schema 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
182 elif schema['type'] == 'tuple': 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
183 if (variadic_item_index := schema.get('variadic_item_index')) is not None: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
184 schema['items_schema'][variadic_item_index] = apply_validators( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
185 schema['items_schema'][variadic_item_index], each_item_validators, field_name
186 )
187 elif is_list_like_schema_with_items_schema(schema): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
188 inner_schema = schema.get('items_schema', None) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
189 if inner_schema is None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
190 inner_schema = core_schema.any_schema() 1eiacnqryszkolmfgtuABCDhjbdpvwExF
191 schema['items_schema'] = apply_validators(inner_schema, each_item_validators, field_name) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
192 elif schema['type'] == 'dict': 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
193 # push down any `each_item=True` validators onto dict _values_
194 # this is super arbitrary but it's the V1 behavior
195 inner_schema = schema.get('values_schema', None) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
196 if inner_schema is None: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
197 inner_schema = core_schema.any_schema() 1eiacnqryszkolmfgtuABCDhjbdpvwExF
198 schema['values_schema'] = apply_validators(inner_schema, each_item_validators, field_name) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
199 elif each_item_validators: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
200 raise TypeError( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
201 f"`@validator(..., each_item=True)` cannot be applied to fields with a schema of {schema['type']}"
202 )
203 return schema 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
206def modify_model_json_schema( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
207 schema_or_field: CoreSchemaOrField, handler: GetJsonSchemaHandler, *, cls: Any
208) -> JsonSchemaValue:
209 """Add title and description for model-like classes' JSON schema.
211 Args:
212 schema_or_field: The schema data to generate a JSON schema from.
213 handler: The `GetCoreSchemaHandler` instance.
214 cls: The model-like class.
216 Returns:
217 JsonSchemaValue: The updated JSON schema.
218 """
219 from ..main import BaseModel 1eiacnqryszkolmfgtuABCDhjbdpvwExF
220 from ..root_model import RootModel 1eiacnqryszkolmfgtuABCDhjbdpvwExF
222 json_schema = handler(schema_or_field) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
223 original_schema = handler.resolve_ref_schema(json_schema) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
224 # Preserve the fact that definitions schemas should never have sibling keys:
225 if '$ref' in original_schema: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
226 ref = original_schema['$ref'] 1eiacnqryszkolmfgtuABCDhjbdpvwExF
227 original_schema.clear() 1eiacnqryszkolmfgtuABCDhjbdpvwExF
228 original_schema['allOf'] = [{'$ref': ref}] 1eiacnqryszkolmfgtuABCDhjbdpvwExF
229 if 'title' not in original_schema: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
230 original_schema['title'] = cls.__name__ 1eiacnqryszkolmfgtuABCDhjbdpvwExF
231 # BaseModel; don't use cls.__doc__ as it will contain the verbose class signature by default
232 docstring = None if cls is BaseModel else cls.__doc__ 1eiacnqryszkolmfgtuABCDhjbdpvwExF
233 if docstring and 'description' not in original_schema: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
234 original_schema['description'] = inspect.cleandoc(docstring) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
235 elif issubclass(cls, RootModel) and cls.model_fields['root'].description: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
236 original_schema['description'] = cls.model_fields['root'].description 1eiacnqryszkolmfgtuABCDhjbdpvwExF
237 return json_schema 1eiacnqryszkolmfgtuABCDhjbdpvwExF
240JsonEncoders = Dict[Type[Any], JsonEncoder] 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
243def _add_custom_serialization_from_json_encoders( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
244 json_encoders: JsonEncoders | None, tp: Any, schema: CoreSchema
245) -> CoreSchema:
246 """Iterate over the json_encoders and add the first matching encoder to the schema.
248 Args:
249 json_encoders: A dictionary of types and their encoder functions.
250 tp: The type to check for a matching encoder.
251 schema: The schema to add the encoder to.
252 """
253 if not json_encoders: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
254 return schema 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
255 if 'serialization' in schema: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
256 return schema 1eiacnqryszkolmfgtuABCDhjbdpvwExF
257 # Check the class type and its superclasses for a matching encoder
258 # Decimal.__class__.__mro__ (and probably other cases) doesn't include Decimal itself
259 # if the type is a GenericAlias (e.g. from list[int]) we need to use __class__ instead of .__mro__
260 for base in (tp, *getattr(tp, '__mro__', tp.__class__.__mro__)[:-1]): 1eiacnqryszkolmfgtuABCDhjbdpvwExF
261 encoder = json_encoders.get(base) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
262 if encoder is None: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
263 continue 1eiacnqryszkolmfgtuABCDhjbdpvwExF
265 warnings.warn( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
266 f'`json_encoders` is deprecated. See https://docs.pydantic.dev/{version_short()}/concepts/serialization/#custom-serializers for alternatives',
267 PydanticDeprecatedSince20,
268 )
270 # TODO: in theory we should check that the schema accepts a serialization key
271 schema['serialization'] = core_schema.plain_serializer_function_ser_schema(encoder, when_used='json') 1eiacnqryszkolmfgtuABCDhjbdpvwExF
272 return schema 1eiacnqryszkolmfgtuABCDhjbdpvwExF
274 return schema 1eiacnqryszkolmfgtuABCDhjbdpvwExF
277TypesNamespace = Union[Dict[str, Any], None] 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
280class TypesNamespaceStack: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
281 """A stack of types namespaces."""
283 def __init__(self, types_namespace: TypesNamespace): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
284 self._types_namespace_stack: list[TypesNamespace] = [types_namespace] 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
286 @property 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
287 def tail(self) -> TypesNamespace: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
288 return self._types_namespace_stack[-1] 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
290 @contextmanager 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
291 def push(self, for_type: type[Any]): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
292 types_namespace = {**_typing_extra.get_cls_types_namespace(for_type), **(self.tail or {})} 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
293 self._types_namespace_stack.append(types_namespace) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
294 try: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
295 yield 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
296 finally:
297 self._types_namespace_stack.pop() 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
300def _get_first_non_null(a: Any, b: Any) -> Any: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
301 """Return the first argument if it is not None, otherwise return the second argument.
303 Use case: serialization_alias (argument a) and alias (argument b) are both defined, and serialization_alias is ''.
304 This function will return serialization_alias, which is the first argument, even though it is an empty string.
305 """
306 return a if a is not None else b 1eiacnqryszkolmfgtuABCDhjbdpvwExF
309class GenerateSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
310 """Generate core schema for a Pydantic model, dataclass and types like `str`, `datetime`, ... ."""
312 __slots__ = ( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
313 '_config_wrapper_stack',
314 '_types_namespace_stack',
315 '_typevars_map',
316 'field_name_stack',
317 'model_type_stack',
318 'defs',
319 )
321 def __init__( 1eiacnqryszlmfgtuABCDGHIJKLMNOhjbdpvwExF
322 self,
323 config_wrapper: ConfigWrapper,
324 types_namespace: dict[str, Any] | None,
325 typevars_map: dict[Any, Any] | None = None,
326 ) -> None:
327 # we need a stack for recursing into child models
328 self._config_wrapper_stack = ConfigWrapperStack(config_wrapper) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
329 self._types_namespace_stack = TypesNamespaceStack(types_namespace) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
330 self._typevars_map = typevars_map 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
331 self.field_name_stack = _FieldNameStack() 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
332 self.model_type_stack = _ModelTypeStack() 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
333 self.defs = _Definitions() 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
335 @classmethod 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
336 def __from_parent( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
337 cls,
338 config_wrapper_stack: ConfigWrapperStack,
339 types_namespace_stack: TypesNamespaceStack,
340 model_type_stack: _ModelTypeStack,
341 typevars_map: dict[Any, Any] | None,
342 defs: _Definitions,
343 ) -> GenerateSchema:
344 obj = cls.__new__(cls) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
345 obj._config_wrapper_stack = config_wrapper_stack 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
346 obj._types_namespace_stack = types_namespace_stack 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
347 obj.model_type_stack = model_type_stack 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
348 obj._typevars_map = typevars_map 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
349 obj.field_name_stack = _FieldNameStack() 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
350 obj.defs = defs 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
351 return obj 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
353 @property 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
354 def _config_wrapper(self) -> ConfigWrapper: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
355 return self._config_wrapper_stack.tail 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
357 @property 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
358 def _types_namespace(self) -> dict[str, Any] | None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
359 return self._types_namespace_stack.tail 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
361 @property 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
362 def _current_generate_schema(self) -> GenerateSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
363 cls = self._config_wrapper.schema_generator or GenerateSchema 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
364 return cls.__from_parent( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
365 self._config_wrapper_stack,
366 self._types_namespace_stack,
367 self.model_type_stack,
368 self._typevars_map,
369 self.defs,
370 )
372 @property 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
373 def _arbitrary_types(self) -> bool: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
374 return self._config_wrapper.arbitrary_types_allowed 1eiacnqryszkolmfgtuABCDhjbdpvwExF
376 def str_schema(self) -> CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
377 """Generate a CoreSchema for `str`"""
378 return core_schema.str_schema() 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
380 # the following methods can be overridden but should be considered
381 # unstable / private APIs
382 def _list_schema(self, tp: Any, items_type: Any) -> CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
383 return core_schema.list_schema(self.generate_schema(items_type)) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
385 def _dict_schema(self, tp: Any, keys_type: Any, values_type: Any) -> CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
386 return core_schema.dict_schema(self.generate_schema(keys_type), self.generate_schema(values_type)) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
388 def _set_schema(self, tp: Any, items_type: Any) -> CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
389 return core_schema.set_schema(self.generate_schema(items_type)) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
391 def _frozenset_schema(self, tp: Any, items_type: Any) -> CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
392 return core_schema.frozenset_schema(self.generate_schema(items_type)) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
394 def _arbitrary_type_schema(self, tp: Any) -> CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
395 if not isinstance(tp, type): 1eiacnqryszkolmfgtuABCDhjbdpvwExF
396 warn( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
397 f'{tp!r} is not a Python type (it may be an instance of an object),'
398 ' Pydantic will allow any object with no validation since we cannot even'
399 ' enforce that the input is an instance of the given type.'
400 ' To get rid of this error wrap the type with `pydantic.SkipValidation`.',
401 UserWarning,
402 )
403 return core_schema.any_schema() 1eiacnqryszkolmfgtuABCDhjbdpvwExF
404 return core_schema.is_instance_schema(tp) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
406 def _unknown_type_schema(self, obj: Any) -> CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
407 raise PydanticSchemaGenerationError( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
408 f'Unable to generate pydantic-core schema for {obj!r}. '
409 'Set `arbitrary_types_allowed=True` in the model_config to ignore this error'
410 ' or implement `__get_pydantic_core_schema__` on your type to fully support it.'
411 '\n\nIf you got this error by calling handler(<some type>) within'
412 ' `__get_pydantic_core_schema__` then you likely need to call'
413 ' `handler.generate_schema(<some type>)` since we do not call'
414 ' `__get_pydantic_core_schema__` on `<some type>` otherwise to avoid infinite recursion.'
415 )
417 def _apply_discriminator_to_union( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
418 self, schema: CoreSchema, discriminator: str | Discriminator | None
419 ) -> CoreSchema:
420 if discriminator is None: 420 ↛ 421line 420 didn't jump to line 421, because the condition on line 420 was never true1eiacnqryszkolmfgtuABCDhjbdpvwExF
421 return schema
422 try: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
423 return _discriminated_union.apply_discriminator( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
424 schema,
425 discriminator,
426 )
427 except _discriminated_union.MissingDefinitionForUnionRef: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
428 # defer until defs are resolved
429 _discriminated_union.set_discriminator_in_metadata( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
430 schema,
431 discriminator,
432 )
433 return schema 1eiacnqryszkolmfgtuABCDhjbdpvwExF
435 class CollectedInvalid(Exception): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
436 pass 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
438 def clean_schema(self, schema: CoreSchema) -> CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
439 schema = self.collect_definitions(schema) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
440 schema = simplify_schema_references(schema) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
441 if collect_invalid_schemas(schema): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
442 raise self.CollectedInvalid() 1eiacnqryszkolmfgtuABCDhjbdpvwExF
443 schema = _discriminated_union.apply_discriminators(schema) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
444 schema = validate_core_schema(schema) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
445 return schema 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
447 def collect_definitions(self, schema: CoreSchema) -> CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
448 ref = cast('str | None', schema.get('ref', None)) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
449 if ref: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
450 self.defs.definitions[ref] = schema 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
451 if 'ref' in schema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
452 schema = core_schema.definition_reference_schema(schema['ref']) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
453 return core_schema.definitions_schema( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
454 schema,
455 list(self.defs.definitions.values()),
456 )
458 def _add_js_function(self, metadata_schema: CoreSchema, js_function: Callable[..., Any]) -> None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
459 metadata = CoreMetadataHandler(metadata_schema).metadata 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
460 pydantic_js_functions = metadata.setdefault('pydantic_js_functions', []) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
461 # because of how we generate core schemas for nested generic models
462 # we can end up adding `BaseModel.__get_pydantic_json_schema__` multiple times
463 # this check may fail to catch duplicates if the function is a `functools.partial`
464 # or something like that
465 # but if it does it'll fail by inserting the duplicate
466 if js_function not in pydantic_js_functions: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
467 pydantic_js_functions.append(js_function) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
469 def generate_schema( 1eiacnqryszlmfgtuABCDGHIJKLMNOhjbdpvwExF
470 self,
471 obj: Any,
472 from_dunder_get_core_schema: bool = True,
473 ) -> core_schema.CoreSchema:
474 """Generate core schema.
476 Args:
477 obj: The object to generate core schema for.
478 from_dunder_get_core_schema: Whether to generate schema from either the
479 `__get_pydantic_core_schema__` function or `__pydantic_core_schema__` property.
481 Returns:
482 The generated core schema.
484 Raises:
485 PydanticUndefinedAnnotation:
486 If it is not possible to evaluate forward reference.
487 PydanticSchemaGenerationError:
488 If it is not possible to generate pydantic-core schema.
489 TypeError:
490 - If `alias_generator` returns a disallowed type (must be str, AliasPath or AliasChoices).
491 - If V1 style validator with `each_item=True` applied on a wrong field.
492 PydanticUserError:
493 - If `typing.TypedDict` is used instead of `typing_extensions.TypedDict` on Python < 3.12.
494 - If `__modify_schema__` method is used instead of `__get_pydantic_json_schema__`.
495 """
496 schema: CoreSchema | None = None 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
498 if from_dunder_get_core_schema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
499 from_property = self._generate_schema_from_property(obj, obj) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
500 if from_property is not None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
501 schema = from_property 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
503 if schema is None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
504 schema = self._generate_schema_inner(obj) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
506 metadata_js_function = _extract_get_pydantic_json_schema(obj, schema) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
507 if metadata_js_function is not None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
508 metadata_schema = resolve_original_schema(schema, self.defs.definitions) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
509 if metadata_schema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
510 self._add_js_function(metadata_schema, metadata_js_function) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
512 schema = _add_custom_serialization_from_json_encoders(self._config_wrapper.json_encoders, obj, schema) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
514 return schema 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
516 def _model_schema(self, cls: type[BaseModel]) -> core_schema.CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
517 """Generate schema for a Pydantic model."""
518 with self.defs.get_schema_or_ref(cls) as (model_ref, maybe_schema): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
519 if maybe_schema is not None: 519 ↛ 520line 519 didn't jump to line 520, because the condition on line 519 was never true1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
520 return maybe_schema
522 fields = cls.model_fields 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
523 decorators = cls.__pydantic_decorators__ 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
524 computed_fields = decorators.computed_fields 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
525 check_decorator_fields_exist( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
526 chain(
527 decorators.field_validators.values(),
528 decorators.field_serializers.values(),
529 decorators.validators.values(),
530 ),
531 {*fields.keys(), *computed_fields.keys()},
532 )
533 config_wrapper = ConfigWrapper(cls.model_config, check=False) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
534 core_config = config_wrapper.core_config(cls) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
535 metadata = build_metadata_dict(js_functions=[partial(modify_model_json_schema, cls=cls)]) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
537 model_validators = decorators.model_validators.values() 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
539 extras_schema = None 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
540 if core_config.get('extra_fields_behavior') == 'allow': 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
541 assert cls.__mro__[0] is cls 1eiacnqryszkolmfgtuABCDhjbdpvwExF
542 assert cls.__mro__[-1] is object 1eiacnqryszkolmfgtuABCDhjbdpvwExF
543 for candidate_cls in cls.__mro__[:-1]: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
544 extras_annotation = getattr(candidate_cls, '__annotations__', {}).get('__pydantic_extra__', None) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
545 if extras_annotation is not None: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
546 if isinstance(extras_annotation, str): 1eiacnqryszkolmfgtuABCDhjbdpvwExF
547 extras_annotation = _typing_extra.eval_type_backport( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
548 _typing_extra._make_forward_ref(extras_annotation, is_argument=False, is_class=True),
549 self._types_namespace,
550 )
551 tp = get_origin(extras_annotation) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
552 if tp not in (Dict, dict): 552 ↛ 553line 552 didn't jump to line 553, because the condition on line 552 was never true1eiacnqryszkolmfgtuABCDhjbdpvwExF
553 raise PydanticSchemaGenerationError(
554 'The type annotation for `__pydantic_extra__` must be `Dict[str, ...]`'
555 )
556 extra_items_type = self._get_args_resolving_forward_refs( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
557 extras_annotation,
558 required=True,
559 )[1]
560 if extra_items_type is not Any: 560 ↛ 543line 560 didn't jump to line 543, because the condition on line 560 was always true1eiacnqryszkolmfgtuABCDhjbdpvwExF
561 extras_schema = self.generate_schema(extra_items_type) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
562 break 1eiacnqryszkolmfgtuABCDhjbdpvwExF
564 with self._config_wrapper_stack.push(config_wrapper), self._types_namespace_stack.push(cls): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
565 self = self._current_generate_schema 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
566 if cls.__pydantic_root_model__: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
567 root_field = self._common_field_schema('root', fields['root'], decorators) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
568 inner_schema = root_field['schema'] 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
569 inner_schema = apply_model_validators(inner_schema, model_validators, 'inner') 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
570 model_schema = core_schema.model_schema( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
571 cls,
572 inner_schema,
573 custom_init=getattr(cls, '__pydantic_custom_init__', None),
574 root_model=True,
575 post_init=getattr(cls, '__pydantic_post_init__', None),
576 config=core_config,
577 ref=model_ref,
578 metadata=metadata,
579 )
580 else:
581 fields_schema: core_schema.CoreSchema = core_schema.model_fields_schema( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
582 {k: self._generate_md_field_schema(k, v, decorators) for k, v in fields.items()},
583 computed_fields=[
584 self._computed_field_schema(d, decorators.field_serializers)
585 for d in computed_fields.values()
586 ],
587 extras_schema=extras_schema,
588 model_name=cls.__name__,
589 )
590 inner_schema = apply_validators(fields_schema, decorators.root_validators.values(), None) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
591 new_inner_schema = define_expected_missing_refs(inner_schema, recursively_defined_type_refs()) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
592 if new_inner_schema is not None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
593 inner_schema = new_inner_schema 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
594 inner_schema = apply_model_validators(inner_schema, model_validators, 'inner') 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
596 model_schema = core_schema.model_schema( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
597 cls,
598 inner_schema,
599 custom_init=getattr(cls, '__pydantic_custom_init__', None),
600 root_model=False,
601 post_init=getattr(cls, '__pydantic_post_init__', None),
602 config=core_config,
603 ref=model_ref,
604 metadata=metadata,
605 )
607 schema = self._apply_model_serializers(model_schema, decorators.model_serializers.values()) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
608 schema = apply_model_validators(schema, model_validators, 'outer') 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
609 self.defs.definitions[model_ref] = schema 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
610 return core_schema.definition_reference_schema(model_ref) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
612 def _unpack_refs_defs(self, schema: CoreSchema) -> CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
613 """Unpack all 'definitions' schemas into `GenerateSchema.defs.definitions`
614 and return the inner schema.
615 """
617 def get_ref(s: CoreSchema) -> str: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
618 return s['ref'] # type: ignore 1eiacnqryszkolmfgtuABCDhjbdpvwExF
620 if schema['type'] == 'definitions': 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
621 self.defs.definitions.update({get_ref(s): s for s in schema['definitions']}) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
622 schema = schema['schema'] 1eiacnqryszkolmfgtuABCDhjbdpvwExF
623 return schema 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
625 def _generate_schema_from_property(self, obj: Any, source: Any) -> core_schema.CoreSchema | None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
626 """Try to generate schema from either the `__get_pydantic_core_schema__` function or
627 `__pydantic_core_schema__` property.
629 Note: `__get_pydantic_core_schema__` takes priority so it can
630 decide whether to use a `__pydantic_core_schema__` attribute, or generate a fresh schema.
631 """
632 # avoid calling `__get_pydantic_core_schema__` if we've already visited this object
633 if is_self_type(obj): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
634 obj = self.model_type_stack.get() 1eiacnqryszkolmfgtuABCDhjbdpvwExF
635 with self.defs.get_schema_or_ref(obj) as (_, maybe_schema): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
636 if maybe_schema is not None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
637 return maybe_schema 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
638 if obj is source: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
639 ref_mode = 'unpack' 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
640 else:
641 ref_mode = 'to-def' 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
643 schema: CoreSchema
645 if (get_schema := getattr(obj, '__get_pydantic_core_schema__', None)) is not None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
646 if len(inspect.signature(get_schema).parameters) == 1: 646 ↛ 648line 646 didn't jump to line 648, because the condition on line 646 was never true1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
647 # (source) -> CoreSchema
648 schema = get_schema(source)
649 else:
650 schema = get_schema( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
651 source, CallbackGetCoreSchemaHandler(self._generate_schema_inner, self, ref_mode=ref_mode)
652 )
653 # fmt: off
654 elif ( 1eiackolmfgGHhjbd
655 (existing_schema := getattr(obj, '__pydantic_core_schema__', None)) is not None
656 and not isinstance(existing_schema, MockCoreSchema)
657 and existing_schema.get('cls', None) == obj
658 ):
659 schema = existing_schema 1eiacnqryszkolmfgtuABCDhjbdpvwExF
660 # fmt: on
661 elif (validators := getattr(obj, '__get_validators__', None)) is not None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
662 warn( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
663 '`__get_validators__` is deprecated and will be removed, use `__get_pydantic_core_schema__` instead.',
664 PydanticDeprecatedSince20,
665 )
666 schema = core_schema.chain_schema([core_schema.with_info_plain_validator_function(v) for v in validators()]) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
667 else:
668 # we have no existing schema information on the property, exit early so that we can go generate a schema
669 return None 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
671 schema = self._unpack_refs_defs(schema) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
673 if is_function_with_inner_schema(schema): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
674 ref = schema['schema'].pop('ref', None) # pyright: ignore[reportCallIssue, reportArgumentType] 1eiacnqryszkolmfgtuABCDhjbdpvwExF
675 if ref: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
676 schema['ref'] = ref 1eiacnqryszkolmfgtuABCDhjbdpvwExF
677 else:
678 ref = get_ref(schema) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
680 if ref: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
681 self.defs.definitions[ref] = schema 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
682 return core_schema.definition_reference_schema(ref) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
684 return schema 1eiacnqryszkolmfgtuABCDhjbdpvwExF
686 def _resolve_forward_ref(self, obj: Any) -> Any: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
687 # we assume that types_namespace has the target of forward references in its scope,
688 # but this could fail, for example, if calling Validator on an imported type which contains
689 # forward references to other types only defined in the module from which it was imported
690 # `Validator(SomeImportedTypeAliasWithAForwardReference)`
691 # or the equivalent for BaseModel
692 # class Model(BaseModel):
693 # x: SomeImportedTypeAliasWithAForwardReference
694 try: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
695 obj = _typing_extra.eval_type_backport(obj, globalns=self._types_namespace) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
696 except NameError as e: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
697 raise PydanticUndefinedAnnotation.from_name_error(e) from e 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
699 # if obj is still a ForwardRef, it means we can't evaluate it, raise PydanticUndefinedAnnotation
700 if isinstance(obj, ForwardRef): 1eiacnqryszkolmfgtuABCDhjbdpvwExF
701 raise PydanticUndefinedAnnotation(obj.__forward_arg__, f'Unable to evaluate forward reference {obj}') 1eiacnqryszkolmfgtuABCDhjbdpvwExF
703 if self._typevars_map: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
704 obj = replace_types(obj, self._typevars_map) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
706 return obj 1eiacnqryszkolmfgtuABCDhjbdpvwExF
708 @overload 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
709 def _get_args_resolving_forward_refs(self, obj: Any, required: Literal[True]) -> tuple[Any, ...]: ... 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
711 @overload 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
712 def _get_args_resolving_forward_refs(self, obj: Any) -> tuple[Any, ...] | None: ... 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
714 def _get_args_resolving_forward_refs(self, obj: Any, required: bool = False) -> tuple[Any, ...] | None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
715 args = get_args(obj) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
716 if args: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
717 args = tuple([self._resolve_forward_ref(a) if isinstance(a, ForwardRef) else a for a in args]) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
718 elif required: # pragma: no cover 1eiacnqryszkolmfgtuABCDhjbdpvwExF
719 raise TypeError(f'Expected {obj} to have generic parameters but it had none')
720 return args 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
722 def _get_first_arg_or_any(self, obj: Any) -> Any: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
723 args = self._get_args_resolving_forward_refs(obj) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
724 if not args: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
725 return Any 1eiacnqryszkolmfgtuABCDhjbdpvwExF
726 return args[0] 1eiacnqryszkolmfgtuABCDhjbdpvwExF
728 def _get_first_two_args_or_any(self, obj: Any) -> tuple[Any, Any]: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
729 args = self._get_args_resolving_forward_refs(obj) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
730 if not args: 730 ↛ 732line 730 didn't jump to line 732, because the condition on line 730 was always true1eiacnqryszkolmfgtuABCDhjbdpvwExF
731 return (Any, Any) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
732 if len(args) < 2:
733 origin = get_origin(obj)
734 raise TypeError(f'Expected two type arguments for {origin}, got 1')
735 return args[0], args[1]
737 def _generate_schema_inner(self, obj: Any) -> core_schema.CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
738 if isinstance(obj, _AnnotatedType): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
739 return self._annotated_schema(obj) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
741 if isinstance(obj, dict): 741 ↛ 743line 741 didn't jump to line 743, because the condition on line 741 was never true1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
742 # we assume this is already a valid schema
743 return obj # type: ignore[return-value]
745 if isinstance(obj, str): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
746 obj = ForwardRef(obj) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
748 if isinstance(obj, ForwardRef): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
749 return self.generate_schema(self._resolve_forward_ref(obj)) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
751 from ..main import BaseModel 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
753 if lenient_issubclass(obj, BaseModel): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
754 with self.model_type_stack.push(obj): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
755 return self._model_schema(obj) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
757 if isinstance(obj, PydanticRecursiveRef): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
758 return core_schema.definition_reference_schema(schema_ref=obj.type_ref) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
760 return self.match_type(obj) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
762 def match_type(self, obj: Any) -> core_schema.CoreSchema: # noqa: C901 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
763 """Main mapping of types to schemas.
765 The general structure is a series of if statements starting with the simple cases
766 (non-generic primitive types) and then handling generics and other more complex cases.
768 Each case either generates a schema directly, calls into a public user-overridable method
769 (like `GenerateSchema.tuple_variable_schema`) or calls into a private method that handles some
770 boilerplate before calling into the user-facing method (e.g. `GenerateSchema._tuple_schema`).
772 The idea is that we'll evolve this into adding more and more user facing methods over time
773 as they get requested and we figure out what the right API for them is.
774 """
775 if obj is str: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
776 return self.str_schema() 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
777 elif obj is bytes: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
778 return core_schema.bytes_schema() 1eiacnqryszkolmfgtuABCDhjbdpvwExF
779 elif obj is int: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
780 return core_schema.int_schema() 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
781 elif obj is float: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
782 return core_schema.float_schema() 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
783 elif obj is bool: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
784 return core_schema.bool_schema() 1eiacnqryszkolmfgtuABCDhjbdpvwExF
785 elif obj is Any or obj is object: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
786 return core_schema.any_schema() 1eiacnqryszkolmfgtuABCDhjbdpvwExF
787 elif obj is None or obj is _typing_extra.NoneType: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
788 return core_schema.none_schema() 1eiacnqryszkolmfgtuABCDhjbdpvwExF
789 elif obj in TUPLE_TYPES: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
790 return self._tuple_schema(obj) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
791 elif obj in LIST_TYPES: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
792 return self._list_schema(obj, self._get_first_arg_or_any(obj)) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
793 elif obj in SET_TYPES: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
794 return self._set_schema(obj, self._get_first_arg_or_any(obj)) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
795 elif obj in FROZEN_SET_TYPES: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
796 return self._frozenset_schema(obj, self._get_first_arg_or_any(obj)) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
797 elif obj in DICT_TYPES: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
798 return self._dict_schema(obj, *self._get_first_two_args_or_any(obj)) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
799 elif isinstance(obj, TypeAliasType): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
800 return self._type_alias_type_schema(obj) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
801 elif obj == type: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
802 return self._type_schema() 1eiacnqryszkolmfgtuABCDhjbdpvwExF
803 elif _typing_extra.is_callable_type(obj): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
804 return core_schema.callable_schema() 1eiacnqryszkolmfgtuABCDhjbdpvwExF
805 elif _typing_extra.is_literal_type(obj): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
806 return self._literal_schema(obj) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
807 elif is_typeddict(obj): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
808 return self._typed_dict_schema(obj, None) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
809 elif _typing_extra.is_namedtuple(obj): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
810 return self._namedtuple_schema(obj, None) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
811 elif _typing_extra.is_new_type(obj): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
812 # NewType, can't use isinstance because it fails <3.10
813 return self.generate_schema(obj.__supertype__) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
814 elif obj == re.Pattern: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
815 return self._pattern_schema(obj) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
816 elif obj is collections.abc.Hashable or obj is typing.Hashable: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
817 return self._hashable_schema() 1eiacnqryszkolmfgtuABCDhjbdpvwExF
818 elif isinstance(obj, typing.TypeVar): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
819 return self._unsubstituted_typevar_schema(obj) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
820 elif is_finalvar(obj): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
821 if obj is Final: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
822 return core_schema.any_schema() 1eiacnqryszkolmfgtuABCDhjbdpvwExF
823 return self.generate_schema( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
824 self._get_first_arg_or_any(obj),
825 )
826 elif isinstance(obj, (FunctionType, LambdaType, MethodType, partial)): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
827 return self._callable_schema(obj) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
828 elif inspect.isclass(obj) and issubclass(obj, Enum): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
829 from ._std_types_schema import get_enum_core_schema 1eiacnqryszkolmfgtuABCDhjbdpvwExF
831 return get_enum_core_schema(obj, self._config_wrapper.config_dict) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
833 if _typing_extra.is_dataclass(obj): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
834 return self._dataclass_schema(obj, None) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
835 res = self._get_prepare_pydantic_annotations_for_known_type(obj, ()) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
836 if res is not None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
837 source_type, annotations = res 1eiacnqryszkolmfgtuABCDhjbdpvwExF
838 return self._apply_annotations(source_type, annotations) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
840 origin = get_origin(obj) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
841 if origin is not None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
842 return self._match_generic_type(obj, origin) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
844 if self._arbitrary_types: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
845 return self._arbitrary_type_schema(obj) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
846 return self._unknown_type_schema(obj) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
848 def _match_generic_type(self, obj: Any, origin: Any) -> CoreSchema: # noqa: C901 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
849 if isinstance(origin, TypeAliasType): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
850 return self._type_alias_type_schema(obj) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
852 # Need to handle generic dataclasses before looking for the schema properties because attribute accesses
853 # on _GenericAlias delegate to the origin type, so lose the information about the concrete parametrization
854 # As a result, currently, there is no way to cache the schema for generic dataclasses. This may be possible
855 # to resolve by modifying the value returned by `Generic.__class_getitem__`, but that is a dangerous game.
856 if _typing_extra.is_dataclass(origin): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
857 return self._dataclass_schema(obj, origin) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
858 if _typing_extra.is_namedtuple(origin): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
859 return self._namedtuple_schema(obj, origin) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
861 from_property = self._generate_schema_from_property(origin, obj) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
862 if from_property is not None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
863 return from_property 1eiacnqryszkolmfgtuABCDhjbdpvwExF
865 if _typing_extra.origin_is_union(origin): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
866 return self._union_schema(obj) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
867 elif origin in TUPLE_TYPES: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
868 return self._tuple_schema(obj) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
869 elif origin in LIST_TYPES: 869 ↛ 870line 869 didn't jump to line 870, because the condition on line 869 was never true1eiacnqryszkolmfgtuABCDhjbdpvwExF
870 return self._list_schema(obj, self._get_first_arg_or_any(obj))
871 elif origin in SET_TYPES: 871 ↛ 872line 871 didn't jump to line 872, because the condition on line 871 was never true1eiacnqryszkolmfgtuABCDhjbdpvwExF
872 return self._set_schema(obj, self._get_first_arg_or_any(obj))
873 elif origin in FROZEN_SET_TYPES: 873 ↛ 874line 873 didn't jump to line 874, because the condition on line 873 was never true1eiacnqryszkolmfgtuABCDhjbdpvwExF
874 return self._frozenset_schema(obj, self._get_first_arg_or_any(obj))
875 elif origin in DICT_TYPES: 875 ↛ 876line 875 didn't jump to line 876, because the condition on line 875 was never true1eiacnqryszkolmfgtuABCDhjbdpvwExF
876 return self._dict_schema(obj, *self._get_first_two_args_or_any(obj))
877 elif is_typeddict(origin): 1eiacnqryszkolmfgtuABCDhjbdpvwExF
878 return self._typed_dict_schema(obj, origin) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
879 elif origin in (typing.Type, type): 1eiacnqryszkolmfgtuABCDhjbdpvwExF
880 return self._subclass_schema(obj) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
881 elif origin in {typing.Sequence, collections.abc.Sequence}: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
882 return self._sequence_schema(obj) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
883 elif origin in {typing.Iterable, collections.abc.Iterable, typing.Generator, collections.abc.Generator}: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
884 return self._iterable_schema(obj) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
885 elif origin in (re.Pattern, typing.Pattern): 1eiacnqryszkolmfgtuABCDhjbdpvwExF
886 return self._pattern_schema(obj) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
888 if self._arbitrary_types: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
889 return self._arbitrary_type_schema(origin) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
890 return self._unknown_type_schema(obj) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
892 def _generate_td_field_schema( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
893 self,
894 name: str,
895 field_info: FieldInfo,
896 decorators: DecoratorInfos,
897 *,
898 required: bool = True,
899 ) -> core_schema.TypedDictField:
900 """Prepare a TypedDictField to represent a model or typeddict field."""
901 common_field = self._common_field_schema(name, field_info, decorators) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
902 return core_schema.typed_dict_field( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
903 common_field['schema'],
904 required=False if not field_info.is_required() else required,
905 serialization_exclude=common_field['serialization_exclude'],
906 validation_alias=common_field['validation_alias'],
907 serialization_alias=common_field['serialization_alias'],
908 metadata=common_field['metadata'],
909 )
911 def _generate_md_field_schema( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
912 self,
913 name: str,
914 field_info: FieldInfo,
915 decorators: DecoratorInfos,
916 ) -> core_schema.ModelField:
917 """Prepare a ModelField to represent a model field."""
918 common_field = self._common_field_schema(name, field_info, decorators) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
919 return core_schema.model_field( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
920 common_field['schema'],
921 serialization_exclude=common_field['serialization_exclude'],
922 validation_alias=common_field['validation_alias'],
923 serialization_alias=common_field['serialization_alias'],
924 frozen=common_field['frozen'],
925 metadata=common_field['metadata'],
926 )
928 def _generate_dc_field_schema( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
929 self,
930 name: str,
931 field_info: FieldInfo,
932 decorators: DecoratorInfos,
933 ) -> core_schema.DataclassField:
934 """Prepare a DataclassField to represent the parameter/field, of a dataclass."""
935 common_field = self._common_field_schema(name, field_info, decorators) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
936 return core_schema.dataclass_field( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
937 name,
938 common_field['schema'],
939 init=field_info.init,
940 init_only=field_info.init_var or None,
941 kw_only=None if field_info.kw_only else False,
942 serialization_exclude=common_field['serialization_exclude'],
943 validation_alias=common_field['validation_alias'],
944 serialization_alias=common_field['serialization_alias'],
945 frozen=common_field['frozen'],
946 metadata=common_field['metadata'],
947 )
949 @staticmethod 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
950 def _apply_alias_generator_to_field_info( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
951 alias_generator: Callable[[str], str] | AliasGenerator, field_info: FieldInfo, field_name: str
952 ) -> None:
953 """Apply an alias_generator to aliases on a FieldInfo instance if appropriate.
955 Args:
956 alias_generator: A callable that takes a string and returns a string, or an AliasGenerator instance.
957 field_info: The FieldInfo instance to which the alias_generator is (maybe) applied.
958 field_name: The name of the field from which to generate the alias.
959 """
960 # Apply an alias_generator if
961 # 1. An alias is not specified
962 # 2. An alias is specified, but the priority is <= 1
963 if ( 1eiackolmfghjbd
964 field_info.alias_priority is None
965 or field_info.alias_priority <= 1
966 or field_info.alias is None
967 or field_info.validation_alias is None
968 or field_info.serialization_alias is None
969 ):
970 alias, validation_alias, serialization_alias = None, None, None 1eiacnqryszkolmfgtuABCDhjbdpvwExF
972 if isinstance(alias_generator, AliasGenerator): 1eiacnqryszkolmfgtuABCDhjbdpvwExF
973 alias, validation_alias, serialization_alias = alias_generator.generate_aliases(field_name) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
974 elif isinstance(alias_generator, Callable): 974 ↛ 982line 974 didn't jump to line 982, because the condition on line 974 was always true1eiacnqryszkolmfgtuABCDhjbdpvwExF
975 alias = alias_generator(field_name) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
976 if not isinstance(alias, str): 1eiacnqryszkolmfgtuABCDhjbdpvwExF
977 raise TypeError(f'alias_generator {alias_generator} must return str, not {alias.__class__}') 1eiacnqryszkolmfgtuABCDhjbdpvwExF
979 # if priority is not set, we set to 1
980 # which supports the case where the alias_generator from a child class is used
981 # to generate an alias for a field in a parent class
982 if field_info.alias_priority is None or field_info.alias_priority <= 1: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
983 field_info.alias_priority = 1 1eiacnqryszkolmfgtuABCDhjbdpvwExF
985 # if the priority is 1, then we set the aliases to the generated alias
986 if field_info.alias_priority == 1: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
987 field_info.serialization_alias = _get_first_non_null(serialization_alias, alias) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
988 field_info.validation_alias = _get_first_non_null(validation_alias, alias) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
989 field_info.alias = alias 1eiacnqryszkolmfgtuABCDhjbdpvwExF
991 # if any of the aliases are not set, then we set them to the corresponding generated alias
992 if field_info.alias is None: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
993 field_info.alias = alias 1eiacnqryszkolmfgtuABCDhjbdpvwExF
994 if field_info.serialization_alias is None: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
995 field_info.serialization_alias = _get_first_non_null(serialization_alias, alias) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
996 if field_info.validation_alias is None: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
997 field_info.validation_alias = _get_first_non_null(validation_alias, alias) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
999 @staticmethod 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1000 def _apply_alias_generator_to_computed_field_info( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1001 alias_generator: Callable[[str], str] | AliasGenerator,
1002 computed_field_info: ComputedFieldInfo,
1003 computed_field_name: str,
1004 ):
1005 """Apply an alias_generator to alias on a ComputedFieldInfo instance if appropriate.
1007 Args:
1008 alias_generator: A callable that takes a string and returns a string, or an AliasGenerator instance.
1009 computed_field_info: The ComputedFieldInfo instance to which the alias_generator is (maybe) applied.
1010 computed_field_name: The name of the computed field from which to generate the alias.
1011 """
1012 # Apply an alias_generator if
1013 # 1. An alias is not specified
1014 # 2. An alias is specified, but the priority is <= 1
1016 if ( 1eiackohjbd
1017 computed_field_info.alias_priority is None
1018 or computed_field_info.alias_priority <= 1
1019 or computed_field_info.alias is None
1020 ):
1021 alias, validation_alias, serialization_alias = None, None, None 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1023 if isinstance(alias_generator, AliasGenerator): 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1024 alias, validation_alias, serialization_alias = alias_generator.generate_aliases(computed_field_name) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1025 elif isinstance(alias_generator, Callable): 1025 ↛ 1033line 1025 didn't jump to line 1033, because the condition on line 1025 was always true1eiacnqryszkolmfgtuABCDhjbdpvwExF
1026 alias = alias_generator(computed_field_name) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1027 if not isinstance(alias, str): 1027 ↛ 1028line 1027 didn't jump to line 1028, because the condition on line 1027 was never true1eiacnqryszkolmfgtuABCDhjbdpvwExF
1028 raise TypeError(f'alias_generator {alias_generator} must return str, not {alias.__class__}')
1030 # if priority is not set, we set to 1
1031 # which supports the case where the alias_generator from a child class is used
1032 # to generate an alias for a field in a parent class
1033 if computed_field_info.alias_priority is None or computed_field_info.alias_priority <= 1: 1033 ↛ 1039line 1033 didn't jump to line 1039, because the condition on line 1033 was always true1eiacnqryszkolmfgtuABCDhjbdpvwExF
1034 computed_field_info.alias_priority = 1 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1036 # if the priority is 1, then we set the aliases to the generated alias
1037 # note that we use the serialization_alias with priority over alias, as computed_field
1038 # aliases are used for serialization only (not validation)
1039 if computed_field_info.alias_priority == 1: 1039 ↛ exitline 1039 didn't return from function '_apply_alias_generator_to_computed_field_info', because the condition on line 1039 was always true1eiacnqryszkolmfgtuABCDhjbdpvwExF
1040 computed_field_info.alias = _get_first_non_null(serialization_alias, alias) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1042 def _common_field_schema( # C901 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1043 self, name: str, field_info: FieldInfo, decorators: DecoratorInfos
1044 ) -> _CommonField:
1045 # Update FieldInfo annotation if appropriate:
1046 from .. import AliasChoices, AliasPath 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1047 from ..fields import FieldInfo 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1049 if has_instance_in_type(field_info.annotation, (ForwardRef, str)): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1050 types_namespace = self._types_namespace 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1051 if self._typevars_map: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1052 types_namespace = (types_namespace or {}).copy() 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1053 # Ensure that typevars get mapped to their concrete types:
1054 types_namespace.update({k.__name__: v for k, v in self._typevars_map.items()}) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1056 evaluated = _typing_extra.eval_type_lenient(field_info.annotation, types_namespace) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1057 if evaluated is not field_info.annotation and not has_instance_in_type(evaluated, PydanticRecursiveRef): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1058 new_field_info = FieldInfo.from_annotation(evaluated) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1059 field_info.annotation = new_field_info.annotation 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1061 # Handle any field info attributes that may have been obtained from now-resolved annotations
1062 for k, v in new_field_info._attributes_set.items(): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1063 # If an attribute is already set, it means it was set by assigning to a call to Field (or just a
1064 # default value), and that should take the highest priority. So don't overwrite existing attributes.
1065 # We skip over "attributes" that are present in the metadata_lookup dict because these won't
1066 # actually end up as attributes of the `FieldInfo` instance.
1067 if k not in field_info._attributes_set and k not in field_info.metadata_lookup: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1068 setattr(field_info, k, v) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1070 # Finally, ensure the field info also reflects all the `_attributes_set` that are actually metadata.
1071 field_info.metadata = [*new_field_info.metadata, *field_info.metadata] 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1073 source_type, annotations = field_info.annotation, field_info.metadata 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1075 def set_discriminator(schema: CoreSchema) -> CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1076 schema = self._apply_discriminator_to_union(schema, field_info.discriminator) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1077 return schema 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1079 with self.field_name_stack.push(name): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1080 if field_info.discriminator is not None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1081 schema = self._apply_annotations(source_type, annotations, transform_inner_schema=set_discriminator) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1082 else:
1083 schema = self._apply_annotations( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1084 source_type,
1085 annotations,
1086 )
1088 # This V1 compatibility shim should eventually be removed
1089 # push down any `each_item=True` validators
1090 # note that this won't work for any Annotated types that get wrapped by a function validator
1091 # but that's okay because that didn't exist in V1
1092 this_field_validators = filter_field_decorator_info_by_field(decorators.validators.values(), name) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1093 if _validators_require_validate_default(this_field_validators): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1094 field_info.validate_default = True 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1095 each_item_validators = [v for v in this_field_validators if v.info.each_item is True] 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1096 this_field_validators = [v for v in this_field_validators if v not in each_item_validators] 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1097 schema = apply_each_item_validators(schema, each_item_validators, name) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1099 schema = apply_validators(schema, filter_field_decorator_info_by_field(this_field_validators, name), name) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1100 schema = apply_validators( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1101 schema, filter_field_decorator_info_by_field(decorators.field_validators.values(), name), name
1102 )
1104 # the default validator needs to go outside of any other validators
1105 # so that it is the topmost validator for the field validator
1106 # which uses it to check if the field has a default value or not
1107 if not field_info.is_required(): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1108 schema = wrap_default(field_info, schema) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1110 schema = self._apply_field_serializers( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1111 schema, filter_field_decorator_info_by_field(decorators.field_serializers.values(), name)
1112 )
1113 json_schema_updates = { 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1114 'title': field_info.title,
1115 'description': field_info.description,
1116 'deprecated': bool(field_info.deprecated) or field_info.deprecated == '' or None,
1117 'examples': to_jsonable_python(field_info.examples),
1118 }
1119 json_schema_updates = {k: v for k, v in json_schema_updates.items() if v is not None} 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1121 json_schema_extra = field_info.json_schema_extra 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1123 metadata = build_metadata_dict( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1124 js_annotation_functions=[get_json_schema_update_func(json_schema_updates, json_schema_extra)]
1125 )
1127 alias_generator = self._config_wrapper.alias_generator 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1128 if alias_generator is not None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1129 self._apply_alias_generator_to_field_info(alias_generator, field_info, name) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1131 if isinstance(field_info.validation_alias, (AliasChoices, AliasPath)): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1132 validation_alias = field_info.validation_alias.convert_to_aliases() 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1133 else:
1134 validation_alias = field_info.validation_alias 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1136 return _common_field( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1137 schema,
1138 serialization_exclude=True if field_info.exclude else None,
1139 validation_alias=validation_alias,
1140 serialization_alias=field_info.serialization_alias,
1141 frozen=field_info.frozen,
1142 metadata=metadata,
1143 )
1145 def _union_schema(self, union_type: Any) -> core_schema.CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1146 """Generate schema for a Union."""
1147 args = self._get_args_resolving_forward_refs(union_type, required=True) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1148 choices: list[CoreSchema] = [] 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1149 nullable = False 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1150 for arg in args: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1151 if arg is None or arg is _typing_extra.NoneType: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1152 nullable = True 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1153 else:
1154 choices.append(self.generate_schema(arg)) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1156 if len(choices) == 1: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1157 s = choices[0] 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1158 else:
1159 choices_with_tags: list[CoreSchema | tuple[CoreSchema, str]] = [] 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1160 for choice in choices: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1161 tag = choice.get('metadata', {}).get(_core_utils.TAGGED_UNION_TAG_KEY) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1162 if tag is not None: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1163 choices_with_tags.append((choice, tag)) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1164 else:
1165 choices_with_tags.append(choice) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1166 s = core_schema.union_schema(choices_with_tags) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1168 if nullable: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1169 s = core_schema.nullable_schema(s) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1170 return s 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1172 def _type_alias_type_schema( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1173 self,
1174 obj: Any, # TypeAliasType
1175 ) -> CoreSchema:
1176 with self.defs.get_schema_or_ref(obj) as (ref, maybe_schema): 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1177 if maybe_schema is not None: 1177 ↛ 1178line 1177 didn't jump to line 1178, because the condition on line 1177 was never true1eiacnqryszkolmfgtuABCDhjbdpvwExF
1178 return maybe_schema
1180 origin = get_origin(obj) or obj 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1182 annotation = origin.__value__ 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1183 typevars_map = get_standard_typevars_map(obj) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1185 with self._types_namespace_stack.push(origin): 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1186 annotation = _typing_extra.eval_type_lenient(annotation, self._types_namespace) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1187 annotation = replace_types(annotation, typevars_map) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1188 schema = self.generate_schema(annotation) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1189 assert schema['type'] != 'definitions' 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1190 schema['ref'] = ref # type: ignore 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1191 self.defs.definitions[ref] = schema 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1192 return core_schema.definition_reference_schema(ref) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1194 def _literal_schema(self, literal_type: Any) -> CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1195 """Generate schema for a Literal."""
1196 expected = _typing_extra.all_literal_values(literal_type) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1197 assert expected, f'literal "expected" cannot be empty, obj={literal_type}' 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1198 return core_schema.literal_schema(expected) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1200 def _typed_dict_schema(self, typed_dict_cls: Any, origin: Any) -> core_schema.CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1201 """Generate schema for a TypedDict.
1203 It is not possible to track required/optional keys in TypedDict without __required_keys__
1204 since TypedDict.__new__ erases the base classes (it replaces them with just `dict`)
1205 and thus we can track usage of total=True/False
1206 __required_keys__ was added in Python 3.9
1207 (https://github.com/miss-islington/cpython/blob/1e9939657dd1f8eb9f596f77c1084d2d351172fc/Doc/library/typing.rst?plain=1#L1546-L1548)
1208 however it is buggy
1209 (https://github.com/python/typing_extensions/blob/ac52ac5f2cb0e00e7988bae1e2a1b8257ac88d6d/src/typing_extensions.py#L657-L666).
1211 On 3.11 but < 3.12 TypedDict does not preserve inheritance information.
1213 Hence to avoid creating validators that do not do what users expect we only
1214 support typing.TypedDict on Python >= 3.12 or typing_extension.TypedDict on all versions
1215 """
1216 from ..fields import FieldInfo 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1218 with self.model_type_stack.push(typed_dict_cls), self.defs.get_schema_or_ref(typed_dict_cls) as ( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1219 typed_dict_ref,
1220 maybe_schema,
1221 ):
1222 if maybe_schema is not None: 1222 ↛ 1223line 1222 didn't jump to line 1223, because the condition on line 1222 was never true1eiacnqryszkolmfgtuABCDhjbdpvwExF
1223 return maybe_schema
1225 typevars_map = get_standard_typevars_map(typed_dict_cls) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1226 if origin is not None: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1227 typed_dict_cls = origin 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1229 if not _SUPPORTS_TYPEDDICT and type(typed_dict_cls).__module__ == 'typing': 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1230 raise PydanticUserError( 1eiacnqrykolmfgtuABhjbdpvwE
1231 'Please use `typing_extensions.TypedDict` instead of `typing.TypedDict` on Python < 3.12.',
1232 code='typed-dict-version',
1233 )
1235 try: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1236 config: ConfigDict | None = get_attribute_from_bases(typed_dict_cls, '__pydantic_config__') 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1237 except AttributeError: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1238 config = None 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1240 with self._config_wrapper_stack.push(config), self._types_namespace_stack.push(typed_dict_cls): 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1241 core_config = self._config_wrapper.core_config(typed_dict_cls) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1243 self = self._current_generate_schema 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1245 required_keys: frozenset[str] = typed_dict_cls.__required_keys__ 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1247 fields: dict[str, core_schema.TypedDictField] = {} 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1249 decorators = DecoratorInfos.build(typed_dict_cls) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1251 if self._config_wrapper.use_attribute_docstrings: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1252 field_docstrings = extract_docstrings_from_cls(typed_dict_cls, use_inspect=True) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1253 else:
1254 field_docstrings = None 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1256 for field_name, annotation in get_type_hints_infer_globalns( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1257 typed_dict_cls, localns=self._types_namespace, include_extras=True
1258 ).items():
1259 annotation = replace_types(annotation, typevars_map) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1260 required = field_name in required_keys 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1262 if get_origin(annotation) == _typing_extra.Required: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1263 required = True 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1264 annotation = self._get_args_resolving_forward_refs( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1265 annotation,
1266 required=True,
1267 )[0]
1268 elif get_origin(annotation) == _typing_extra.NotRequired: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1269 required = False 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1270 annotation = self._get_args_resolving_forward_refs( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1271 annotation,
1272 required=True,
1273 )[0]
1275 field_info = FieldInfo.from_annotation(annotation) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1276 if ( 1eiackolmfghjbd
1277 field_docstrings is not None
1278 and field_info.description is None
1279 and field_name in field_docstrings
1280 ):
1281 field_info.description = field_docstrings[field_name] 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1282 fields[field_name] = self._generate_td_field_schema( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1283 field_name, field_info, decorators, required=required
1284 )
1286 metadata = build_metadata_dict( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1287 js_functions=[partial(modify_model_json_schema, cls=typed_dict_cls)], typed_dict_cls=typed_dict_cls
1288 )
1290 td_schema = core_schema.typed_dict_schema( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1291 fields,
1292 computed_fields=[
1293 self._computed_field_schema(d, decorators.field_serializers)
1294 for d in decorators.computed_fields.values()
1295 ],
1296 ref=typed_dict_ref,
1297 metadata=metadata,
1298 config=core_config,
1299 )
1301 schema = self._apply_model_serializers(td_schema, decorators.model_serializers.values()) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1302 schema = apply_model_validators(schema, decorators.model_validators.values(), 'all') 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1303 self.defs.definitions[typed_dict_ref] = schema 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1304 return core_schema.definition_reference_schema(typed_dict_ref) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1306 def _namedtuple_schema(self, namedtuple_cls: Any, origin: Any) -> core_schema.CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1307 """Generate schema for a NamedTuple."""
1308 with self.model_type_stack.push(namedtuple_cls), self.defs.get_schema_or_ref(namedtuple_cls) as ( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1309 namedtuple_ref,
1310 maybe_schema,
1311 ):
1312 if maybe_schema is not None: 1312 ↛ 1313line 1312 didn't jump to line 1313, because the condition on line 1312 was never true1eiacnqryszkolmfgtuABCDhjbdpvwExF
1313 return maybe_schema
1314 typevars_map = get_standard_typevars_map(namedtuple_cls) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1315 if origin is not None: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1316 namedtuple_cls = origin 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1318 annotations: dict[str, Any] = get_type_hints_infer_globalns( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1319 namedtuple_cls, include_extras=True, localns=self._types_namespace
1320 )
1321 if not annotations: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1322 # annotations is empty, happens if namedtuple_cls defined via collections.namedtuple(...)
1323 annotations = {k: Any for k in namedtuple_cls._fields} 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1325 if typevars_map: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1326 annotations = { 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1327 field_name: replace_types(annotation, typevars_map)
1328 for field_name, annotation in annotations.items()
1329 }
1331 arguments_schema = core_schema.arguments_schema( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1332 [
1333 self._generate_parameter_schema(
1334 field_name, annotation, default=namedtuple_cls._field_defaults.get(field_name, Parameter.empty)
1335 )
1336 for field_name, annotation in annotations.items()
1337 ],
1338 metadata=build_metadata_dict(js_prefer_positional_arguments=True),
1339 )
1340 return core_schema.call_schema(arguments_schema, namedtuple_cls, ref=namedtuple_ref) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1342 def _generate_parameter_schema( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1343 self,
1344 name: str,
1345 annotation: type[Any],
1346 default: Any = Parameter.empty,
1347 mode: Literal['positional_only', 'positional_or_keyword', 'keyword_only'] | None = None,
1348 ) -> core_schema.ArgumentsParameter:
1349 """Prepare a ArgumentsParameter to represent a field in a namedtuple or function signature."""
1350 from ..fields import FieldInfo 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1352 if default is Parameter.empty: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1353 field = FieldInfo.from_annotation(annotation) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1354 else:
1355 field = FieldInfo.from_annotated_attribute(annotation, default) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1356 assert field.annotation is not None, 'field.annotation should not be None when generating a schema' 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1357 source_type, annotations = field.annotation, field.metadata 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1358 with self.field_name_stack.push(name): 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1359 schema = self._apply_annotations(source_type, annotations) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1361 if not field.is_required(): 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1362 schema = wrap_default(field, schema) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1364 parameter_schema = core_schema.arguments_parameter(name, schema) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1365 if mode is not None: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1366 parameter_schema['mode'] = mode 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1367 if field.alias is not None: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1368 parameter_schema['alias'] = field.alias 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1369 else:
1370 alias_generator = self._config_wrapper.alias_generator 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1371 if isinstance(alias_generator, AliasGenerator) and alias_generator.alias is not None: 1371 ↛ 1372line 1371 didn't jump to line 1372, because the condition on line 1371 was never true1eiacnqryszkolmfgtuABCDhjbdpvwExF
1372 parameter_schema['alias'] = alias_generator.alias(name)
1373 elif isinstance(alias_generator, Callable): 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1374 parameter_schema['alias'] = alias_generator(name) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1375 return parameter_schema 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1377 def _tuple_schema(self, tuple_type: Any) -> core_schema.CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1378 """Generate schema for a Tuple, e.g. `tuple[int, str]` or `tuple[int, ...]`."""
1379 # TODO: do we really need to resolve type vars here?
1380 typevars_map = get_standard_typevars_map(tuple_type) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1381 params = self._get_args_resolving_forward_refs(tuple_type) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1383 if typevars_map and params: 1383 ↛ 1384line 1383 didn't jump to line 1384, because the condition on line 1383 was never true1eiacnqryszkolmfgtuABCDhjbdpvwExF
1384 params = tuple(replace_types(param, typevars_map) for param in params)
1386 # NOTE: subtle difference: `tuple[()]` gives `params=()`, whereas `typing.Tuple[()]` gives `params=((),)`
1387 # This is only true for <3.11, on Python 3.11+ `typing.Tuple[()]` gives `params=()`
1388 if not params: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1389 if tuple_type in TUPLE_TYPES: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1390 return core_schema.tuple_schema([core_schema.any_schema()], variadic_item_index=0) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1391 else:
1392 # special case for `tuple[()]` which means `tuple[]` - an empty tuple
1393 return core_schema.tuple_schema([]) 1acnqryszkofgtuABCDbdpvwExF
1394 elif params[-1] is Ellipsis: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1395 if len(params) == 2: 1395 ↛ 1399line 1395 didn't jump to line 1399, because the condition on line 1395 was always true1eiacnqryszkolmfgtuABCDhjbdpvwExF
1396 return core_schema.tuple_schema([self.generate_schema(params[0])], variadic_item_index=0) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1397 else:
1398 # TODO: something like https://github.com/pydantic/pydantic/issues/5952
1399 raise ValueError('Variable tuples can only have one type')
1400 elif len(params) == 1 and params[0] == (): 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1401 # special case for `Tuple[()]` which means `Tuple[]` - an empty tuple
1402 # NOTE: This conditional can be removed when we drop support for Python 3.10.
1403 return core_schema.tuple_schema([]) 1eiacnqkolmfgtuhjbdpv
1404 else:
1405 return core_schema.tuple_schema([self.generate_schema(param) for param in params]) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1407 def _type_schema(self) -> core_schema.CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1408 return core_schema.custom_error_schema( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1409 core_schema.is_instance_schema(type),
1410 custom_error_type='is_type',
1411 custom_error_message='Input should be a type',
1412 )
1414 def _union_is_subclass_schema(self, union_type: Any) -> core_schema.CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1415 """Generate schema for `Type[Union[X, ...]]`."""
1416 args = self._get_args_resolving_forward_refs(union_type, required=True) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1417 return core_schema.union_schema([self.generate_schema(typing.Type[args]) for args in args]) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1419 def _subclass_schema(self, type_: Any) -> core_schema.CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1420 """Generate schema for a Type, e.g. `Type[int]`."""
1421 type_param = self._get_first_arg_or_any(type_) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1422 if type_param == Any: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1423 return self._type_schema() 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1424 elif isinstance(type_param, typing.TypeVar): 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1425 if type_param.__bound__: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1426 if _typing_extra.origin_is_union(get_origin(type_param.__bound__)): 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1427 return self._union_is_subclass_schema(type_param.__bound__) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1428 return core_schema.is_subclass_schema(type_param.__bound__) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1429 elif type_param.__constraints__: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1430 return core_schema.union_schema( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1431 [self.generate_schema(typing.Type[c]) for c in type_param.__constraints__]
1432 )
1433 else:
1434 return self._type_schema() 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1435 elif _typing_extra.origin_is_union(get_origin(type_param)): 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1436 return self._union_is_subclass_schema(type_param) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1437 else:
1438 return core_schema.is_subclass_schema(type_param) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1440 def _sequence_schema(self, sequence_type: Any) -> core_schema.CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1441 """Generate schema for a Sequence, e.g. `Sequence[int]`."""
1442 from ._std_types_schema import serialize_sequence_via_list 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1444 item_type = self._get_first_arg_or_any(sequence_type) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1445 item_type_schema = self.generate_schema(item_type) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1446 list_schema = core_schema.list_schema(item_type_schema) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1448 python_schema = core_schema.is_instance_schema(typing.Sequence, cls_repr='Sequence') 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1449 if item_type != Any: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1450 from ._validators import sequence_validator 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1452 python_schema = core_schema.chain_schema( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1453 [python_schema, core_schema.no_info_wrap_validator_function(sequence_validator, list_schema)],
1454 )
1456 serialization = core_schema.wrap_serializer_function_ser_schema( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1457 serialize_sequence_via_list, schema=item_type_schema, info_arg=True
1458 )
1459 return core_schema.json_or_python_schema( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1460 json_schema=list_schema, python_schema=python_schema, serialization=serialization
1461 )
1463 def _iterable_schema(self, type_: Any) -> core_schema.GeneratorSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1464 """Generate a schema for an `Iterable`."""
1465 item_type = self._get_first_arg_or_any(type_) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1467 return core_schema.generator_schema(self.generate_schema(item_type)) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1469 def _pattern_schema(self, pattern_type: Any) -> core_schema.CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1470 from . import _validators 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1472 metadata = build_metadata_dict(js_functions=[lambda _1, _2: {'type': 'string', 'format': 'regex'}]) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1473 ser = core_schema.plain_serializer_function_ser_schema( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1474 attrgetter('pattern'), when_used='json', return_schema=core_schema.str_schema()
1475 )
1476 if pattern_type == typing.Pattern or pattern_type == re.Pattern: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1477 # bare type
1478 return core_schema.no_info_plain_validator_function( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1479 _validators.pattern_either_validator, serialization=ser, metadata=metadata
1480 )
1482 param = self._get_args_resolving_forward_refs( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1483 pattern_type,
1484 required=True,
1485 )[0]
1486 if param == str: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1487 return core_schema.no_info_plain_validator_function( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1488 _validators.pattern_str_validator, serialization=ser, metadata=metadata
1489 )
1490 elif param == bytes: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1491 return core_schema.no_info_plain_validator_function( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1492 _validators.pattern_bytes_validator, serialization=ser, metadata=metadata
1493 )
1494 else:
1495 raise PydanticSchemaGenerationError(f'Unable to generate pydantic-core schema for {pattern_type!r}.') 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1497 def _hashable_schema(self) -> core_schema.CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1498 return core_schema.custom_error_schema( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1499 core_schema.is_instance_schema(collections.abc.Hashable),
1500 custom_error_type='is_hashable',
1501 custom_error_message='Input should be hashable',
1502 )
1504 def _dataclass_schema( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1505 self, dataclass: type[StandardDataclass], origin: type[StandardDataclass] | None
1506 ) -> core_schema.CoreSchema:
1507 """Generate schema for a dataclass."""
1508 with self.model_type_stack.push(dataclass), self.defs.get_schema_or_ref(dataclass) as ( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1509 dataclass_ref,
1510 maybe_schema,
1511 ):
1512 if maybe_schema is not None: 1512 ↛ 1513line 1512 didn't jump to line 1513, because the condition on line 1512 was never true1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1513 return maybe_schema
1515 typevars_map = get_standard_typevars_map(dataclass) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1516 if origin is not None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1517 dataclass = origin 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1519 with ExitStack() as dataclass_bases_stack: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1520 # Pushing a namespace prioritises items already in the stack, so iterate though the MRO forwards
1521 for dataclass_base in dataclass.__mro__: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1522 if dataclasses.is_dataclass(dataclass_base): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1523 dataclass_bases_stack.enter_context(self._types_namespace_stack.push(dataclass_base)) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1525 # Pushing a config overwrites the previous config, so iterate though the MRO backwards
1526 for dataclass_base in reversed(dataclass.__mro__): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1527 if dataclasses.is_dataclass(dataclass_base): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1528 config = getattr(dataclass_base, '__pydantic_config__', None) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1529 dataclass_bases_stack.enter_context(self._config_wrapper_stack.push(config)) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1531 core_config = self._config_wrapper.core_config(dataclass) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1533 self = self._current_generate_schema 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1535 from ..dataclasses import is_pydantic_dataclass 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1537 if is_pydantic_dataclass(dataclass): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1538 fields = deepcopy(dataclass.__pydantic_fields__) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1539 if typevars_map: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1540 for field in fields.values(): 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1541 field.apply_typevars_map(typevars_map, self._types_namespace) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1542 else:
1543 fields = collect_dataclass_fields( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1544 dataclass,
1545 self._types_namespace,
1546 typevars_map=typevars_map,
1547 )
1549 # disallow combination of init=False on a dataclass field and extra='allow' on a dataclass
1550 if self._config_wrapper_stack.tail.extra == 'allow': 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1551 # disallow combination of init=False on a dataclass field and extra='allow' on a dataclass
1552 for field_name, field in fields.items(): 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1553 if field.init is False: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1554 raise PydanticUserError( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1555 f'Field {field_name} has `init=False` and dataclass has config setting `extra="allow"`. '
1556 f'This combination is not allowed.',
1557 code='dataclass-init-false-extra-allow',
1558 )
1560 decorators = dataclass.__dict__.get('__pydantic_decorators__') or DecoratorInfos.build(dataclass) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1561 # Move kw_only=False args to the start of the list, as this is how vanilla dataclasses work.
1562 # Note that when kw_only is missing or None, it is treated as equivalent to kw_only=True
1563 args = sorted( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1564 (self._generate_dc_field_schema(k, v, decorators) for k, v in fields.items()),
1565 key=lambda a: a.get('kw_only') is not False,
1566 )
1567 has_post_init = hasattr(dataclass, '__post_init__') 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1568 has_slots = hasattr(dataclass, '__slots__') 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1570 args_schema = core_schema.dataclass_args_schema( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1571 dataclass.__name__,
1572 args,
1573 computed_fields=[
1574 self._computed_field_schema(d, decorators.field_serializers)
1575 for d in decorators.computed_fields.values()
1576 ],
1577 collect_init_only=has_post_init,
1578 )
1580 inner_schema = apply_validators(args_schema, decorators.root_validators.values(), None) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1582 model_validators = decorators.model_validators.values() 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1583 inner_schema = apply_model_validators(inner_schema, model_validators, 'inner') 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1585 dc_schema = core_schema.dataclass_schema( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1586 dataclass,
1587 inner_schema,
1588 post_init=has_post_init,
1589 ref=dataclass_ref,
1590 fields=[field.name for field in dataclasses.fields(dataclass)],
1591 slots=has_slots,
1592 config=core_config,
1593 )
1594 schema = self._apply_model_serializers(dc_schema, decorators.model_serializers.values()) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1595 schema = apply_model_validators(schema, model_validators, 'outer') 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1596 self.defs.definitions[dataclass_ref] = schema 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1597 return core_schema.definition_reference_schema(dataclass_ref) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1599 # Type checkers seem to assume ExitStack may suppress exceptions and therefore
1600 # control flow can exit the `with` block without returning.
1601 assert False, 'Unreachable' 1ackfgbd
1603 def _callable_schema(self, function: Callable[..., Any]) -> core_schema.CallSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1604 """Generate schema for a Callable.
1606 TODO support functional validators once we support them in Config
1607 """
1608 sig = signature(function) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1610 type_hints = _typing_extra.get_function_type_hints(function) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1612 mode_lookup: dict[_ParameterKind, Literal['positional_only', 'positional_or_keyword', 'keyword_only']] = { 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1613 Parameter.POSITIONAL_ONLY: 'positional_only',
1614 Parameter.POSITIONAL_OR_KEYWORD: 'positional_or_keyword',
1615 Parameter.KEYWORD_ONLY: 'keyword_only',
1616 }
1618 arguments_list: list[core_schema.ArgumentsParameter] = [] 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1619 var_args_schema: core_schema.CoreSchema | None = None 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1620 var_kwargs_schema: core_schema.CoreSchema | None = None 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1622 for name, p in sig.parameters.items(): 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1623 if p.annotation is sig.empty: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1624 annotation = typing.cast(Any, Any) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1625 else:
1626 annotation = type_hints[name] 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1628 parameter_mode = mode_lookup.get(p.kind) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1629 if parameter_mode is not None: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1630 arg_schema = self._generate_parameter_schema(name, annotation, p.default, parameter_mode) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1631 arguments_list.append(arg_schema) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1632 elif p.kind == Parameter.VAR_POSITIONAL: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1633 var_args_schema = self.generate_schema(annotation) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1634 else:
1635 assert p.kind == Parameter.VAR_KEYWORD, p.kind 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1636 var_kwargs_schema = self.generate_schema(annotation) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1638 return_schema: core_schema.CoreSchema | None = None 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1639 config_wrapper = self._config_wrapper 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1640 if config_wrapper.validate_return: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1641 return_hint = type_hints.get('return') 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1642 if return_hint is not None: 1642 ↛ 1645line 1642 didn't jump to line 1645, because the condition on line 1642 was always true1eiacnqryszkolmfgtuABCDhjbdpvwExF
1643 return_schema = self.generate_schema(return_hint) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1645 return core_schema.call_schema( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1646 core_schema.arguments_schema(
1647 arguments_list,
1648 var_args_schema=var_args_schema,
1649 var_kwargs_schema=var_kwargs_schema,
1650 populate_by_name=config_wrapper.populate_by_name,
1651 ),
1652 function,
1653 return_schema=return_schema,
1654 )
1656 def _unsubstituted_typevar_schema(self, typevar: typing.TypeVar) -> core_schema.CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1657 assert isinstance(typevar, typing.TypeVar) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1659 bound = typevar.__bound__ 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1660 constraints = typevar.__constraints__ 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1662 try: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1663 typevar_has_default = typevar.has_default() # type: ignore 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1664 except AttributeError: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1665 # could still have a default if it's an old version of typing_extensions.TypeVar
1666 typevar_has_default = getattr(typevar, '__default__', None) is not None 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1668 if (bound is not None) + (len(constraints) != 0) + typevar_has_default > 1: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1669 raise NotImplementedError( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1670 'Pydantic does not support mixing more than one of TypeVar bounds, constraints and defaults'
1671 )
1673 if typevar_has_default: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1674 return self.generate_schema(typevar.__default__) # type: ignore 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1675 elif constraints: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1676 return self._union_schema(typing.Union[constraints]) # type: ignore 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1677 elif bound: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1678 schema = self.generate_schema(bound) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1679 schema['serialization'] = core_schema.wrap_serializer_function_ser_schema( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1680 lambda x, h: h(x), schema=core_schema.any_schema()
1681 )
1682 return schema 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1683 else:
1684 return core_schema.any_schema() 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1686 def _computed_field_schema( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1687 self,
1688 d: Decorator[ComputedFieldInfo],
1689 field_serializers: dict[str, Decorator[FieldSerializerDecoratorInfo]],
1690 ) -> core_schema.ComputedField:
1691 try: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1692 return_type = _decorators.get_function_return_type(d.func, d.info.return_type, self._types_namespace) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1693 except NameError as e: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1694 raise PydanticUndefinedAnnotation.from_name_error(e) from e 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1695 if return_type is PydanticUndefined: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1696 raise PydanticUserError( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1697 'Computed field is missing return type annotation or specifying `return_type`'
1698 ' to the `@computed_field` decorator (e.g. `@computed_field(return_type=int|str)`)',
1699 code='model-field-missing-annotation',
1700 )
1702 return_type = replace_types(return_type, self._typevars_map) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1703 # Create a new ComputedFieldInfo so that different type parametrizations of the same
1704 # generic model's computed field can have different return types.
1705 d.info = dataclasses.replace(d.info, return_type=return_type) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1706 return_type_schema = self.generate_schema(return_type) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1707 # Apply serializers to computed field if there exist
1708 return_type_schema = self._apply_field_serializers( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1709 return_type_schema,
1710 filter_field_decorator_info_by_field(field_serializers.values(), d.cls_var_name),
1711 computed_field=True,
1712 )
1714 alias_generator = self._config_wrapper.alias_generator 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1715 if alias_generator is not None: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1716 self._apply_alias_generator_to_computed_field_info( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1717 alias_generator=alias_generator, computed_field_info=d.info, computed_field_name=d.cls_var_name
1718 )
1720 def set_computed_field_metadata(schema: CoreSchemaOrField, handler: GetJsonSchemaHandler) -> JsonSchemaValue: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1721 json_schema = handler(schema) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1723 json_schema['readOnly'] = True 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1725 title = d.info.title 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1726 if title is not None: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1727 json_schema['title'] = title 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1729 description = d.info.description 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1730 if description is not None: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1731 json_schema['description'] = description 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1733 if d.info.deprecated or d.info.deprecated == '': 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1734 json_schema['deprecated'] = True 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1736 examples = d.info.examples 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1737 if examples is not None: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1738 json_schema['examples'] = to_jsonable_python(examples) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1740 json_schema_extra = d.info.json_schema_extra 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1741 if json_schema_extra is not None: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1742 add_json_schema_extra(json_schema, json_schema_extra) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1744 return json_schema 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1746 metadata = build_metadata_dict(js_annotation_functions=[set_computed_field_metadata]) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1747 return core_schema.computed_field( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1748 d.cls_var_name, return_schema=return_type_schema, alias=d.info.alias, metadata=metadata
1749 )
1751 def _annotated_schema(self, annotated_type: Any) -> core_schema.CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1752 """Generate schema for an Annotated type, e.g. `Annotated[int, Field(...)]` or `Annotated[int, Gt(0)]`."""
1753 from ..fields import FieldInfo 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1755 source_type, *annotations = self._get_args_resolving_forward_refs( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1756 annotated_type,
1757 required=True,
1758 )
1759 schema = self._apply_annotations(source_type, annotations) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1760 # put the default validator last so that TypeAdapter.get_default_value() works
1761 # even if there are function validators involved
1762 for annotation in annotations: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1763 if isinstance(annotation, FieldInfo): 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1764 schema = wrap_default(annotation, schema) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1765 return schema 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1767 def _get_prepare_pydantic_annotations_for_known_type( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1768 self, obj: Any, annotations: tuple[Any, ...]
1769 ) -> tuple[Any, list[Any]] | None:
1770 from ._std_types_schema import PREPARE_METHODS 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1772 # Check for hashability
1773 try: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1774 hash(obj) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1775 except TypeError: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1776 # obj is definitely not a known type if this fails
1777 return None 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1779 for gen in PREPARE_METHODS: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1780 res = gen(obj, annotations, self._config_wrapper.config_dict) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1781 if res is not None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1782 return res 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1784 return None 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1786 def _apply_annotations( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1787 self,
1788 source_type: Any,
1789 annotations: list[Any],
1790 transform_inner_schema: Callable[[CoreSchema], CoreSchema] = lambda x: x,
1791 ) -> CoreSchema:
1792 """Apply arguments from `Annotated` or from `FieldInfo` to a schema.
1794 This gets called by `GenerateSchema._annotated_schema` but differs from it in that it does
1795 not expect `source_type` to be an `Annotated` object, it expects it to be the first argument of that
1796 (in other words, `GenerateSchema._annotated_schema` just unpacks `Annotated`, this process it).
1797 """
1798 annotations = list(_known_annotated_metadata.expand_grouped_metadata(annotations)) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1799 res = self._get_prepare_pydantic_annotations_for_known_type(source_type, tuple(annotations)) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1800 if res is not None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1801 source_type, annotations = res 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1803 pydantic_js_annotation_functions: list[GetJsonSchemaFunction] = [] 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1805 def inner_handler(obj: Any) -> CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1806 from_property = self._generate_schema_from_property(obj, source_type) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1807 if from_property is None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1808 schema = self._generate_schema_inner(obj) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1809 else:
1810 schema = from_property 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1811 metadata_js_function = _extract_get_pydantic_json_schema(obj, schema) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1812 if metadata_js_function is not None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1813 metadata_schema = resolve_original_schema(schema, self.defs.definitions) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1814 if metadata_schema is not None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1815 self._add_js_function(metadata_schema, metadata_js_function) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1816 return transform_inner_schema(schema) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1818 get_inner_schema = CallbackGetCoreSchemaHandler(inner_handler, self) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1820 for annotation in annotations: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1821 if annotation is None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1822 continue 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1823 get_inner_schema = self._get_wrapped_inner_schema( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1824 get_inner_schema, annotation, pydantic_js_annotation_functions
1825 )
1827 schema = get_inner_schema(source_type) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1828 if pydantic_js_annotation_functions: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1829 metadata = CoreMetadataHandler(schema).metadata 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1830 metadata.setdefault('pydantic_js_annotation_functions', []).extend(pydantic_js_annotation_functions) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1831 return _add_custom_serialization_from_json_encoders(self._config_wrapper.json_encoders, source_type, schema) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1833 def _apply_single_annotation(self, schema: core_schema.CoreSchema, metadata: Any) -> core_schema.CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1834 from ..fields import FieldInfo 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1836 if isinstance(metadata, FieldInfo): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1837 for field_metadata in metadata.metadata: 1837 ↛ 1838line 1837 didn't jump to line 1838, because the loop on line 1837 never started1eiacnqryszkolmfgtuABCDhjbdpvwExF
1838 schema = self._apply_single_annotation(schema, field_metadata)
1840 if metadata.discriminator is not None: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1841 schema = self._apply_discriminator_to_union(schema, metadata.discriminator) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1842 return schema 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1844 if schema['type'] == 'nullable': 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1845 # for nullable schemas, metadata is automatically applied to the inner schema
1846 inner = schema.get('schema', core_schema.any_schema()) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1847 inner = self._apply_single_annotation(inner, metadata) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1848 if inner: 1848 ↛ 1850line 1848 didn't jump to line 1850, because the condition on line 1848 was always true1eiacnqryszkolmfgtuABCDhjbdpvwExF
1849 schema['schema'] = inner 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1850 return schema 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1852 original_schema = schema 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1853 ref = schema.get('ref', None) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1854 if ref is not None: 1854 ↛ 1855line 1854 didn't jump to line 1855, because the condition on line 1854 was never true1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1855 schema = schema.copy()
1856 new_ref = ref + f'_{repr(metadata)}'
1857 if new_ref in self.defs.definitions:
1858 return self.defs.definitions[new_ref]
1859 schema['ref'] = new_ref # type: ignore
1860 elif schema['type'] == 'definition-ref': 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1861 ref = schema['schema_ref'] 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1862 if ref in self.defs.definitions: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1863 schema = self.defs.definitions[ref].copy() 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1864 new_ref = ref + f'_{repr(metadata)}' 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1865 if new_ref in self.defs.definitions: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1866 return self.defs.definitions[new_ref] 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1867 schema['ref'] = new_ref # type: ignore 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1869 maybe_updated_schema = _known_annotated_metadata.apply_known_metadata(metadata, schema.copy()) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1871 if maybe_updated_schema is not None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1872 return maybe_updated_schema 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1873 return original_schema 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1875 def _apply_single_annotation_json_schema( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1876 self, schema: core_schema.CoreSchema, metadata: Any
1877 ) -> core_schema.CoreSchema:
1878 from ..fields import FieldInfo 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1880 if isinstance(metadata, FieldInfo): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1881 for field_metadata in metadata.metadata: 1881 ↛ 1882line 1881 didn't jump to line 1882, because the loop on line 1881 never started1eiacnqryszkolmfgtuABCDhjbdpvwExF
1882 schema = self._apply_single_annotation_json_schema(schema, field_metadata)
1883 json_schema_update: JsonSchemaValue = {} 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1884 if metadata.title: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1885 json_schema_update['title'] = metadata.title 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1886 if metadata.description: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1887 json_schema_update['description'] = metadata.description 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1888 if metadata.examples: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1889 json_schema_update['examples'] = to_jsonable_python(metadata.examples) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1891 json_schema_extra = metadata.json_schema_extra 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1892 if json_schema_update or json_schema_extra: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1893 CoreMetadataHandler(schema).metadata.setdefault('pydantic_js_annotation_functions', []).append( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1894 get_json_schema_update_func(json_schema_update, json_schema_extra)
1895 )
1896 return schema 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1898 def _get_wrapped_inner_schema( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1899 self,
1900 get_inner_schema: GetCoreSchemaHandler,
1901 annotation: Any,
1902 pydantic_js_annotation_functions: list[GetJsonSchemaFunction],
1903 ) -> CallbackGetCoreSchemaHandler:
1904 metadata_get_schema: GetCoreSchemaFunction = getattr(annotation, '__get_pydantic_core_schema__', None) or ( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1905 lambda source, handler: handler(source)
1906 )
1908 def new_handler(source: Any) -> core_schema.CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1909 schema = metadata_get_schema(source, get_inner_schema) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1910 schema = self._apply_single_annotation(schema, annotation) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1911 schema = self._apply_single_annotation_json_schema(schema, annotation) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1913 metadata_js_function = _extract_get_pydantic_json_schema(annotation, schema) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1914 if metadata_js_function is not None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1915 pydantic_js_annotation_functions.append(metadata_js_function) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1916 return schema 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1918 return CallbackGetCoreSchemaHandler(new_handler, self) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1920 def _apply_field_serializers( 1eiacnqryszlmfgtuABCDGHIJKLMNOhjbdpvwExF
1921 self,
1922 schema: core_schema.CoreSchema,
1923 serializers: list[Decorator[FieldSerializerDecoratorInfo]],
1924 computed_field: bool = False,
1925 ) -> core_schema.CoreSchema:
1926 """Apply field serializers to a schema."""
1927 if serializers: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1928 schema = copy(schema) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1929 if schema['type'] == 'definitions': 1929 ↛ 1930line 1929 didn't jump to line 1930, because the condition on line 1929 was never true1eiacnqryszkolmfgtuABCDhjbdpvwExF
1930 inner_schema = schema['schema']
1931 schema['schema'] = self._apply_field_serializers(inner_schema, serializers)
1932 return schema
1933 else:
1934 ref = typing.cast('str|None', schema.get('ref', None)) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1935 if ref is not None: 1935 ↛ 1936line 1935 didn't jump to line 1936, because the condition on line 1935 was never true1eiacnqryszkolmfgtuABCDhjbdpvwExF
1936 schema = core_schema.definition_reference_schema(ref)
1938 # use the last serializer to make it easy to override a serializer set on a parent model
1939 serializer = serializers[-1] 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1940 is_field_serializer, info_arg = inspect_field_serializer( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1941 serializer.func, serializer.info.mode, computed_field=computed_field
1942 )
1944 try: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1945 return_type = _decorators.get_function_return_type( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1946 serializer.func, serializer.info.return_type, self._types_namespace
1947 )
1948 except NameError as e: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1949 raise PydanticUndefinedAnnotation.from_name_error(e) from e 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1951 if return_type is PydanticUndefined: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1952 return_schema = None 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1953 else:
1954 return_schema = self.generate_schema(return_type) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1956 if serializer.info.mode == 'wrap': 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1957 schema['serialization'] = core_schema.wrap_serializer_function_ser_schema( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1958 serializer.func,
1959 is_field_serializer=is_field_serializer,
1960 info_arg=info_arg,
1961 return_schema=return_schema,
1962 when_used=serializer.info.when_used,
1963 )
1964 else:
1965 assert serializer.info.mode == 'plain' 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1966 schema['serialization'] = core_schema.plain_serializer_function_ser_schema( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1967 serializer.func,
1968 is_field_serializer=is_field_serializer,
1969 info_arg=info_arg,
1970 return_schema=return_schema,
1971 when_used=serializer.info.when_used,
1972 )
1973 return schema 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1975 def _apply_model_serializers( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1976 self, schema: core_schema.CoreSchema, serializers: Iterable[Decorator[ModelSerializerDecoratorInfo]]
1977 ) -> core_schema.CoreSchema:
1978 """Apply model serializers to a schema."""
1979 ref: str | None = schema.pop('ref', None) # type: ignore 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1980 if serializers: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
1981 serializer = list(serializers)[-1] 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1982 info_arg = inspect_model_serializer(serializer.func, serializer.info.mode) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1984 try: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1985 return_type = _decorators.get_function_return_type( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1986 serializer.func, serializer.info.return_type, self._types_namespace
1987 )
1988 except NameError as e:
1989 raise PydanticUndefinedAnnotation.from_name_error(e) from e
1990 if return_type is PydanticUndefined: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1991 return_schema = None 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1992 else:
1993 return_schema = self.generate_schema(return_type) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1995 if serializer.info.mode == 'wrap': 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1996 ser_schema: core_schema.SerSchema = core_schema.wrap_serializer_function_ser_schema( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
1997 serializer.func,
1998 info_arg=info_arg,
1999 return_schema=return_schema,
2000 when_used=serializer.info.when_used,
2001 )
2002 else:
2003 # plain
2004 ser_schema = core_schema.plain_serializer_function_ser_schema( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2005 serializer.func,
2006 info_arg=info_arg,
2007 return_schema=return_schema,
2008 when_used=serializer.info.when_used,
2009 )
2010 schema['serialization'] = ser_schema 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2011 if ref: 2011 ↛ 2013line 2011 didn't jump to line 2013, because the condition on line 2011 was always true1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2012 schema['ref'] = ref # type: ignore 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2013 return schema 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2016_VALIDATOR_F_MATCH: Mapping[ 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2017 tuple[FieldValidatorModes, Literal['no-info', 'with-info']],
2018 Callable[[Callable[..., Any], core_schema.CoreSchema, str | None], core_schema.CoreSchema],
2019] = {
2020 ('before', 'no-info'): lambda f, schema, _: core_schema.no_info_before_validator_function(f, schema),
2021 ('after', 'no-info'): lambda f, schema, _: core_schema.no_info_after_validator_function(f, schema),
2022 ('plain', 'no-info'): lambda f, _1, _2: core_schema.no_info_plain_validator_function(f),
2023 ('wrap', 'no-info'): lambda f, schema, _: core_schema.no_info_wrap_validator_function(f, schema),
2024 ('before', 'with-info'): lambda f, schema, field_name: core_schema.with_info_before_validator_function(
2025 f, schema, field_name=field_name
2026 ),
2027 ('after', 'with-info'): lambda f, schema, field_name: core_schema.with_info_after_validator_function(
2028 f, schema, field_name=field_name
2029 ),
2030 ('plain', 'with-info'): lambda f, _, field_name: core_schema.with_info_plain_validator_function(
2031 f, field_name=field_name
2032 ),
2033 ('wrap', 'with-info'): lambda f, schema, field_name: core_schema.with_info_wrap_validator_function(
2034 f, schema, field_name=field_name
2035 ),
2036}
2039def apply_validators( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2040 schema: core_schema.CoreSchema,
2041 validators: Iterable[Decorator[RootValidatorDecoratorInfo]]
2042 | Iterable[Decorator[ValidatorDecoratorInfo]]
2043 | Iterable[Decorator[FieldValidatorDecoratorInfo]],
2044 field_name: str | None,
2045) -> core_schema.CoreSchema:
2046 """Apply validators to a schema.
2048 Args:
2049 schema: The schema to apply validators on.
2050 validators: An iterable of validators.
2051 field_name: The name of the field if validators are being applied to a model field.
2053 Returns:
2054 The updated schema.
2055 """
2056 for validator in validators: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2057 info_arg = inspect_validator(validator.func, validator.info.mode) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2058 val_type = 'with-info' if info_arg else 'no-info' 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2060 schema = _VALIDATOR_F_MATCH[(validator.info.mode, val_type)](validator.func, schema, field_name) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2061 return schema 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2064def _validators_require_validate_default(validators: Iterable[Decorator[ValidatorDecoratorInfo]]) -> bool: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2065 """In v1, if any of the validators for a field had `always=True`, the default value would be validated.
2067 This serves as an auxiliary function for re-implementing that logic, by looping over a provided
2068 collection of (v1-style) ValidatorDecoratorInfo's and checking if any of them have `always=True`.
2070 We should be able to drop this function and the associated logic calling it once we drop support
2071 for v1-style validator decorators. (Or we can extend it and keep it if we add something equivalent
2072 to the v1-validator `always` kwarg to `field_validator`.)
2073 """
2074 for validator in validators: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2075 if validator.info.always: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2076 return True 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2077 return False 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2080def apply_model_validators( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2081 schema: core_schema.CoreSchema,
2082 validators: Iterable[Decorator[ModelValidatorDecoratorInfo]],
2083 mode: Literal['inner', 'outer', 'all'],
2084) -> core_schema.CoreSchema:
2085 """Apply model validators to a schema.
2087 If mode == 'inner', only "before" validators are applied
2088 If mode == 'outer', validators other than "before" are applied
2089 If mode == 'all', all validators are applied
2091 Args:
2092 schema: The schema to apply validators on.
2093 validators: An iterable of validators.
2094 mode: The validator mode.
2096 Returns:
2097 The updated schema.
2098 """
2099 ref: str | None = schema.pop('ref', None) # type: ignore 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2100 for validator in validators: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2101 if mode == 'inner' and validator.info.mode != 'before': 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2102 continue 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2103 if mode == 'outer' and validator.info.mode == 'before': 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2104 continue 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2105 info_arg = inspect_validator(validator.func, validator.info.mode) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2106 if validator.info.mode == 'wrap': 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2107 if info_arg: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2108 schema = core_schema.with_info_wrap_validator_function(function=validator.func, schema=schema) 1szCDxF
2109 else:
2110 schema = core_schema.no_info_wrap_validator_function(function=validator.func, schema=schema) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2111 elif validator.info.mode == 'before': 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2112 if info_arg: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2113 schema = core_schema.with_info_before_validator_function(function=validator.func, schema=schema) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2114 else:
2115 schema = core_schema.no_info_before_validator_function(function=validator.func, schema=schema) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2116 else:
2117 assert validator.info.mode == 'after' 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2118 if info_arg: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2119 schema = core_schema.with_info_after_validator_function(function=validator.func, schema=schema) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2120 else:
2121 schema = core_schema.no_info_after_validator_function(function=validator.func, schema=schema) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2122 if ref: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2123 schema['ref'] = ref # type: ignore 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2124 return schema 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2127def wrap_default(field_info: FieldInfo, schema: core_schema.CoreSchema) -> core_schema.CoreSchema: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2128 """Wrap schema with default schema if default value or `default_factory` are available.
2130 Args:
2131 field_info: The field info object.
2132 schema: The schema to apply default on.
2134 Returns:
2135 Updated schema by default value or `default_factory`.
2136 """
2137 if field_info.default_factory: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2138 return core_schema.with_default_schema( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2139 schema, default_factory=field_info.default_factory, validate_default=field_info.validate_default
2140 )
2141 elif field_info.default is not PydanticUndefined: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2142 return core_schema.with_default_schema( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2143 schema, default=field_info.default, validate_default=field_info.validate_default
2144 )
2145 else:
2146 return schema 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2149def _extract_get_pydantic_json_schema(tp: Any, schema: CoreSchema) -> GetJsonSchemaFunction | None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2150 """Extract `__get_pydantic_json_schema__` from a type, handling the deprecated `__modify_schema__`."""
2151 js_modify_function = getattr(tp, '__get_pydantic_json_schema__', None) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2153 if hasattr(tp, '__modify_schema__'): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2154 from pydantic import BaseModel # circular reference 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2156 has_custom_v2_modify_js_func = ( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2157 js_modify_function is not None
2158 and BaseModel.__get_pydantic_json_schema__.__func__ # type: ignore
2159 not in (js_modify_function, getattr(js_modify_function, '__func__', None))
2160 )
2162 if not has_custom_v2_modify_js_func: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2163 cls_name = getattr(tp, '__name__', None) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2164 raise PydanticUserError( 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2165 f'The `__modify_schema__` method is not supported in Pydantic v2. '
2166 f'Use `__get_pydantic_json_schema__` instead{f" in class `{cls_name}`" if cls_name else ""}.',
2167 code='custom-json-schema',
2168 )
2170 # handle GenericAlias' but ignore Annotated which "lies" about its origin (in this case it would be `int`)
2171 if hasattr(tp, '__origin__') and not isinstance(tp, type(Annotated[int, 'placeholder'])): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2172 return _extract_get_pydantic_json_schema(tp.__origin__, schema) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2174 if js_modify_function is None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2175 return None 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2177 return js_modify_function 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2180def get_json_schema_update_func( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2181 json_schema_update: JsonSchemaValue, json_schema_extra: JsonDict | typing.Callable[[JsonDict], None] | None
2182) -> GetJsonSchemaFunction:
2183 def json_schema_update_func( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2184 core_schema_or_field: CoreSchemaOrField, handler: GetJsonSchemaHandler
2185 ) -> JsonSchemaValue:
2186 json_schema = {**handler(core_schema_or_field), **json_schema_update} 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2187 add_json_schema_extra(json_schema, json_schema_extra) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2188 return json_schema 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2190 return json_schema_update_func 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2193def add_json_schema_extra( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2194 json_schema: JsonSchemaValue, json_schema_extra: JsonDict | typing.Callable[[JsonDict], None] | None
2195):
2196 if isinstance(json_schema_extra, dict): 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2197 json_schema.update(to_jsonable_python(json_schema_extra)) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2198 elif callable(json_schema_extra): 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2199 json_schema_extra(json_schema) 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2202class _CommonField(TypedDict): 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2203 schema: core_schema.CoreSchema 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2204 validation_alias: str | list[str | int] | list[list[str | int]] | None 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2205 serialization_alias: str | None 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2206 serialization_exclude: bool | None 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2207 frozen: bool | None 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2208 metadata: dict[str, Any] 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2211def _common_field( 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2212 schema: core_schema.CoreSchema,
2213 *,
2214 validation_alias: str | list[str | int] | list[list[str | int]] | None = None,
2215 serialization_alias: str | None = None,
2216 serialization_exclude: bool | None = None,
2217 frozen: bool | None = None,
2218 metadata: Any = None,
2219) -> _CommonField:
2220 return { 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2221 'schema': schema,
2222 'validation_alias': validation_alias,
2223 'serialization_alias': serialization_alias,
2224 'serialization_exclude': serialization_exclude,
2225 'frozen': frozen,
2226 'metadata': metadata,
2227 }
2230class _Definitions: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2231 """Keeps track of references and definitions."""
2233 def __init__(self) -> None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2234 self.seen: set[str] = set() 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2235 self.definitions: dict[str, core_schema.CoreSchema] = {} 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2237 @contextmanager 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2238 def get_schema_or_ref(self, tp: Any) -> Iterator[tuple[str, None] | tuple[str, CoreSchema]]: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2239 """Get a definition for `tp` if one exists.
2241 If a definition exists, a tuple of `(ref_string, CoreSchema)` is returned.
2242 If no definition exists yet, a tuple of `(ref_string, None)` is returned.
2244 Note that the returned `CoreSchema` will always be a `DefinitionReferenceSchema`,
2245 not the actual definition itself.
2247 This should be called for any type that can be identified by reference.
2248 This includes any recursive types.
2250 At present the following types can be named/recursive:
2252 - BaseModel
2253 - Dataclasses
2254 - TypedDict
2255 - TypeAliasType
2256 """
2257 ref = get_type_ref(tp) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2258 # return the reference if we're either (1) in a cycle or (2) it was already defined
2259 if ref in self.seen or ref in self.definitions: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2260 yield (ref, core_schema.definition_reference_schema(ref)) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2261 else:
2262 self.seen.add(ref) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2263 try: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2264 yield (ref, None) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2265 finally:
2266 self.seen.discard(ref) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2269def resolve_original_schema(schema: CoreSchema, definitions: dict[str, CoreSchema]) -> CoreSchema | None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2270 if schema['type'] == 'definition-ref': 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2271 return definitions.get(schema['schema_ref'], None) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2272 elif schema['type'] == 'definitions': 2272 ↛ 2273line 2272 didn't jump to line 2273, because the condition on line 2272 was never true1eiacnqryszkolmfgtuABCDhjbdpvwExF
2273 return schema['schema']
2274 else:
2275 return schema 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2278class _FieldNameStack: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2279 __slots__ = ('_stack',) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2281 def __init__(self) -> None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2282 self._stack: list[str] = [] 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2284 @contextmanager 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2285 def push(self, field_name: str) -> Iterator[None]: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2286 self._stack.append(field_name) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2287 yield 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2288 self._stack.pop() 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2290 def get(self) -> str | None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2291 if self._stack: 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2292 return self._stack[-1] 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2293 else:
2294 return None 1eanrshbpwx
2297class _ModelTypeStack: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2298 __slots__ = ('_stack',) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2300 def __init__(self) -> None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2301 self._stack: list[type] = [] 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2303 @contextmanager 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2304 def push(self, type_obj: type) -> Iterator[None]: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2305 self._stack.append(type_obj) 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2306 yield 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2307 self._stack.pop() 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2309 def get(self) -> type | None: 1eiacnqryszkolmfgtuABCDGHIJKLMNOhjbdpvwExF
2310 if self._stack: 2310 ↛ 2313line 2310 didn't jump to line 2313, because the condition on line 2310 was always true1eiacnqryszkolmfgtuABCDhjbdpvwExF
2311 return self._stack[-1] 1eiacnqryszkolmfgtuABCDhjbdpvwExF
2312 else:
2313 return None