Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import operator
4import sys
5import types
6import typing
7from collections import ChainMap
8from collections.abc import Generator, Mapping
9from contextlib import contextmanager
10from contextvars import ContextVar
11from functools import reduce
12from itertools import zip_longest
13from types import prepare_class
14from typing import TYPE_CHECKING, Annotated, Any, TypedDict, TypeVar, cast
15from weakref import WeakValueDictionary
17import typing_extensions
18from typing_inspection import typing_objects
19from typing_inspection.introspection import is_union_origin
21from . import _type_refs, _typing_extra
22from ._forward_ref import PydanticRecursiveRef
23from ._utils import all_identical, is_model_class
25if TYPE_CHECKING:
26 from ..main import BaseModel
28GenericTypesCacheKey = tuple[Any, Any, tuple[Any, ...]]
30# Note: We want to remove LimitedDict, but to do this, we'd need to improve the handling of generics caching.
31# Right now, to handle recursive generics, we some types must remain cached for brief periods without references.
32# By chaining the WeakValuesDict with a LimitedDict, we have a way to retain caching for all types with references,
33# while also retaining a limited number of types even without references. This is generally enough to build
34# specific recursive generic models without losing required items out of the cache.
36KT = TypeVar('KT')
37VT = TypeVar('VT')
38_LIMITED_DICT_SIZE = 100
41class LimitedDict(dict[KT, VT]):
42 def __init__(self, size_limit: int = _LIMITED_DICT_SIZE) -> None:
43 self.size_limit = size_limit
44 super().__init__()
46 def __setitem__(self, key: KT, value: VT, /) -> None:
47 super().__setitem__(key, value)
48 if len(self) > self.size_limit:
49 excess = len(self) - self.size_limit + self.size_limit // 10
50 to_remove = list(self.keys())[:excess]
51 for k in to_remove:
52 del self[k]
55# weak dictionaries allow the dynamically created parametrized versions of generic models to get collected
56# once they are no longer referenced by the caller.
57GenericTypesCache = WeakValueDictionary[GenericTypesCacheKey, 'type[BaseModel]']
59if TYPE_CHECKING:
61 class DeepChainMap(ChainMap[KT, VT]): # type: ignore
62 ...
64else:
66 class DeepChainMap(ChainMap):
67 """Variant of ChainMap that allows direct updates to inner scopes.
69 Taken from https://docs.python.org/3/library/collections.html#collections.ChainMap,
70 with some light modifications for this use case.
71 """
73 def clear(self) -> None:
74 for mapping in self.maps:
75 mapping.clear()
77 def __setitem__(self, key: KT, value: VT) -> None:
78 for mapping in self.maps:
79 mapping[key] = value
81 def __delitem__(self, key: KT) -> None:
82 hit = False
83 for mapping in self.maps:
84 if key in mapping:
85 del mapping[key]
86 hit = True
87 if not hit:
88 raise KeyError(key)
91# Despite the fact that LimitedDict _seems_ no longer necessary, I'm very nervous to actually remove it
92# and discover later on that we need to re-add all this infrastructure...
93# _GENERIC_TYPES_CACHE = DeepChainMap(GenericTypesCache(), LimitedDict())
95_GENERIC_TYPES_CACHE = GenericTypesCache()
98class PydanticGenericMetadata(TypedDict):
99 origin: type[BaseModel] | None # analogous to typing._GenericAlias.__origin__
100 args: tuple[Any, ...] # analogous to typing._GenericAlias.__args__
101 parameters: tuple[TypeVar, ...] # analogous to typing.Generic.__parameters__
104def create_generic_submodel(
105 model_name: str, origin: type[BaseModel], args: tuple[Any, ...], params: tuple[Any, ...]
106) -> type[BaseModel]:
107 """Dynamically create a submodel of a provided (generic) BaseModel.
109 This is used when producing concrete parametrizations of generic models. This function
110 only *creates* the new subclass; the schema/validators/serialization must be updated to
111 reflect a concrete parametrization elsewhere.
113 Args:
114 model_name: The name of the newly created model.
115 origin: The base class for the new model to inherit from.
116 args: A tuple of generic metadata arguments.
117 params: A tuple of generic metadata parameters.
119 Returns:
120 The created submodel.
121 """
122 namespace: dict[str, Any] = {'__module__': origin.__module__}
123 # As per https://docs.python.org/3/reference/datamodel.html#slots:
124 # "The action of a __slots__ declaration is not limited to the class where it is defined.
125 # __slots__ declared in parents are available in child classes. However, instances of a
126 # child subclass will get a __dict__ and __weakref__ unless the subclass also defines
127 # __slots__".
128 # Because when users parameterize a generic model, we create a subclass of such generic model
129 # (what happens in this function), they can't control the fact that no slots is defined on the
130 # dynamic subclass, and so even if they defined extra __slots__ on the generic class (which results
131 # in the class *not* being weakref-able), the parameterized class *will* be weakref-able.
132 # For this reason (and to make Pydantic generic models behavior closer to generic aliases),
133 # we forward any slots from the origin:
134 if '__slots__' in origin.__dict__:
135 namespace['__slots__'] = origin.__dict__['__slots__']
136 bases = (origin,)
137 meta, ns, kwds = prepare_class(model_name, bases)
138 namespace.update(ns)
139 created_model = meta(
140 model_name,
141 bases,
142 namespace,
143 __pydantic_generic_metadata__={
144 'origin': origin,
145 'args': args,
146 'parameters': params,
147 },
148 __pydantic_reset_parent_namespace__=False,
149 **kwds,
150 )
152 model_module, called_globally = _get_caller_frame_info(depth=3)
153 if called_globally: # create global reference and therefore allow pickling
154 object_by_reference = None
155 reference_name = model_name
156 reference_module_globals = sys.modules[model_module or created_model.__module__].__dict__
157 while object_by_reference is not created_model:
158 object_by_reference = reference_module_globals.setdefault(reference_name, created_model)
159 reference_name += '_'
161 return created_model
164def _get_caller_frame_info(depth: int = 2) -> tuple[str | None, bool]:
165 """Used inside a function to check whether it was called globally.
167 Args:
168 depth: The depth to get the frame.
170 Returns:
171 A tuple contains `module_name` and `called_globally`.
173 Raises:
174 RuntimeError: If the function is not called inside a function.
175 """
176 try:
177 previous_caller_frame = sys._getframe(depth)
178 except ValueError as e:
179 raise RuntimeError('This function must be used inside another function') from e
180 except AttributeError: # sys module does not have _getframe function, so there's nothing we can do about it
181 return None, False
182 frame_globals = previous_caller_frame.f_globals
183 return frame_globals.get('__name__'), previous_caller_frame.f_locals is frame_globals
186DictValues: type[Any] = {}.values().__class__
189def iter_contained_typevars(v: Any) -> Generator[TypeVar]:
190 """Recursively iterate through all subtypes and type args of `v` and yield any typevars that are found.
192 This is inspired as an alternative to directly accessing the `__parameters__` attribute of a GenericAlias,
193 since __parameters__ of (nested) generic BaseModel subclasses won't show up in that list.
194 """
195 if isinstance(v, TypeVar):
196 yield v
197 elif is_model_class(v):
198 yield from v.__pydantic_generic_metadata__['parameters']
199 elif isinstance(v, (DictValues, list)):
200 for var in v:
201 yield from iter_contained_typevars(var)
202 else:
203 args = get_args(v)
204 for arg in args:
205 yield from iter_contained_typevars(arg)
208def get_args(v: Any) -> Any:
209 pydantic_generic_metadata: PydanticGenericMetadata | None = getattr(v, '__pydantic_generic_metadata__', None)
210 if pydantic_generic_metadata:
211 return pydantic_generic_metadata.get('args')
212 return typing_extensions.get_args(v)
215def get_origin(v: Any) -> Any:
216 pydantic_generic_metadata: PydanticGenericMetadata | None = getattr(v, '__pydantic_generic_metadata__', None)
217 if pydantic_generic_metadata:
218 return pydantic_generic_metadata.get('origin')
219 return typing_extensions.get_origin(v)
222def get_standard_typevars_map(cls: Any) -> dict[TypeVar, Any] | None:
223 """Package a generic type's typevars and parametrization (if present) into a dictionary compatible with the
224 `replace_types` function. Specifically, this works with standard typing generics and typing._GenericAlias.
225 """
226 origin = get_origin(cls)
227 if origin is None:
228 return None
229 if not hasattr(origin, '__parameters__'):
230 return None
232 # In this case, we know that cls is a _GenericAlias, and origin is the generic type
233 # So it is safe to access cls.__args__ and origin.__parameters__
234 args: tuple[Any, ...] = cls.__args__ # type: ignore
235 parameters: tuple[TypeVar, ...] = origin.__parameters__
236 return dict(zip(parameters, args, strict=True))
239def get_model_typevars_map(cls: type[BaseModel]) -> dict[TypeVar, Any]:
240 """Package a generic BaseModel's typevars and concrete parametrization (if present) into a dictionary compatible
241 with the `replace_types` function.
243 Since BaseModel.__class_getitem__ does not produce a typing._GenericAlias, and the BaseModel generic info is
244 stored in the __pydantic_generic_metadata__ attribute, we need special handling here.
245 """
246 # TODO: This could be unified with `get_standard_typevars_map` if we stored the generic metadata
247 # in the __origin__, __args__, and __parameters__ attributes of the model.
248 generic_metadata = cls.__pydantic_generic_metadata__
249 origin = generic_metadata['origin']
250 args = generic_metadata['args']
251 if not args:
252 # No need to go into `iter_contained_typevars`:
253 return {}
254 return dict(zip(iter_contained_typevars(origin), args, strict=True))
257def replace_types(type_: Any, type_map: Mapping[TypeVar, Any] | None) -> Any:
258 """Return type with all occurrences of `type_map` keys recursively replaced with their values.
260 Args:
261 type_: The class or generic alias.
262 type_map: Mapping from `TypeVar` instance to concrete types.
264 Returns:
265 A new type representing the basic structure of `type_` with all
266 `typevar_map` keys recursively replaced.
268 Example:
269 ```python
270 from pydantic._internal._generics import replace_types
272 replace_types(tuple[str, list[str] | float], {str: int})
273 #> tuple[int, list[int] | float]
274 ```
275 """
276 if not type_map:
277 return type_
279 type_args = get_args(type_)
280 origin_type = get_origin(type_)
282 if typing_objects.is_annotated(origin_type):
283 annotated_type, *annotations = type_args
284 annotated_type = replace_types(annotated_type, type_map)
285 # TODO remove parentheses when we drop support for Python 3.10:
286 return Annotated[(annotated_type, *annotations)]
288 # Having type args is a good indicator that this is a typing special form
289 # instance or a generic alias of some sort.
290 if type_args:
291 resolved_type_args = tuple(replace_types(arg, type_map) for arg in type_args)
292 if all_identical(type_args, resolved_type_args):
293 # If all arguments are the same, there is no need to modify the
294 # type or create a new object at all
295 return type_
297 if (
298 origin_type is not None
299 and isinstance(type_, _typing_extra.typing_base)
300 and not isinstance(origin_type, _typing_extra.typing_base)
301 and getattr(type_, '_name', None) is not None
302 ):
303 # In python < 3.9 generic aliases don't exist so any of these like `list`,
304 # `type` or `collections.abc.Callable` need to be translated.
305 # See: https://www.python.org/dev/peps/pep-0585
306 origin_type = getattr(typing, type_._name)
307 assert origin_type is not None
309 if is_union_origin(origin_type):
310 if any(typing_objects.is_any(arg) for arg in resolved_type_args):
311 # `Any | T` ~ `Any`:
312 resolved_type_args = (Any,)
313 # `Never | T` ~ `T`:
314 resolved_type_args = tuple(
315 arg
316 for arg in resolved_type_args
317 if not (typing_objects.is_noreturn(arg) or typing_objects.is_never(arg))
318 )
320 # PEP-604 syntax (e.g. `list | str`) is represented with a types.UnionType object that does not
321 # implement `__getitem__()`. In Python 3.14+, `typing.Union` and `types.UnionType` are the same,
322 # and we instead rely on `typing.Union` as it implicitly converts string annotations to `ForwardRef`
323 # instances (this is to avoid type errors as per https://github.com/python/cpython/pull/105366).
324 if sys.version_info < (3, 14) and origin_type is types.UnionType:
325 return reduce(operator.or_, resolved_type_args)
326 # NotRequired[T] and Required[T] don't support tuple type resolved_type_args, hence the condition below
327 return origin_type[resolved_type_args[0] if len(resolved_type_args) == 1 else resolved_type_args]
329 # We handle pydantic generic models separately as they don't have the same
330 # semantics as "typing" classes or generic aliases
332 if not origin_type and is_model_class(type_):
333 parameters = type_.__pydantic_generic_metadata__['parameters']
334 if not parameters:
335 return type_
336 resolved_type_args = tuple(replace_types(t, type_map) for t in parameters)
337 if all_identical(parameters, resolved_type_args):
338 return type_
339 return type_[resolved_type_args]
341 # Handle special case for typehints that can have lists as arguments.
342 # `typing.Callable[[int, str], int]` is an example for this.
343 if isinstance(type_, list):
344 resolved_list = [replace_types(element, type_map) for element in type_]
345 if all_identical(type_, resolved_list):
346 return type_
347 return resolved_list
349 # If all else fails, we try to resolve the type directly and otherwise just
350 # return the input with no modifications.
351 return type_map.get(type_, type_)
354def map_generic_model_arguments(cls: type[BaseModel], args: tuple[Any, ...]) -> dict[TypeVar, Any]:
355 """Return a mapping between the parameters of a generic model and the provided arguments during parameterization.
357 Raises:
358 TypeError: If the number of arguments does not match the parameters (i.e. if providing too few or too many arguments).
360 Example:
361 ```python {test="skip" lint="skip"}
362 class Model[T, U, V = int](BaseModel): ...
364 map_generic_model_arguments(Model, (str, bytes))
365 #> {T: str, U: bytes, V: int}
367 map_generic_model_arguments(Model, (str,))
368 #> TypeError: Too few arguments for <class '__main__.Model'>; actual 1, expected at least 2
370 map_generic_model_arguments(Model, (str, bytes, int, complex))
371 #> TypeError: Too many arguments for <class '__main__.Model'>; actual 4, expected 3
372 ```
374 Note:
375 This function is analogous to the private `typing._check_generic_specialization` function.
376 """
377 parameters = cls.__pydantic_generic_metadata__['parameters']
378 expected_len = len(parameters)
379 typevars_map: dict[TypeVar, Any] = {}
381 _missing = object()
382 for parameter, argument in zip_longest(parameters, args, fillvalue=_missing):
383 if parameter is _missing:
384 raise TypeError(f'Too many arguments for {cls}; actual {len(args)}, expected {expected_len}')
386 if argument is _missing:
387 param = cast(TypeVar, parameter)
388 try:
389 has_default = param.has_default() # pyright: ignore[reportAttributeAccessIssue]
390 except AttributeError:
391 # Happens if using `typing.TypeVar` (and not `typing_extensions`) on Python < 3.13.
392 has_default = False
393 if has_default:
394 # The default might refer to other type parameters. For an example, see:
395 # https://typing.python.org/en/latest/spec/generics.html#type-parameters-as-parameters-to-generics
396 typevars_map[param] = replace_types(param.__default__, typevars_map) # pyright: ignore[reportAttributeAccessIssue]
397 else:
398 expected_len -= sum(hasattr(p, 'has_default') and p.has_default() for p in parameters) # pyright: ignore[reportAttributeAccessIssue]
399 raise TypeError(f'Too few arguments for {cls}; actual {len(args)}, expected at least {expected_len}')
400 else:
401 param = cast(TypeVar, parameter)
402 typevars_map[param] = argument
404 return typevars_map
407_generic_recursion_cache: ContextVar[set[str] | None] = ContextVar('_generic_recursion_cache', default=None)
410@contextmanager
411def generic_recursion_self_type(
412 origin: type[BaseModel], args: tuple[Any, ...]
413) -> Generator[PydanticRecursiveRef | None]:
414 """This contextmanager should be placed around the recursive calls used to build a generic type,
415 and accept as arguments the generic origin type and the type arguments being passed to it.
417 If the same origin and arguments are observed twice, it implies that a self-reference placeholder
418 can be used while building the core schema, and will produce a schema_ref that will be valid in the
419 final parent schema.
420 """
421 previously_seen_type_refs = _generic_recursion_cache.get()
422 if previously_seen_type_refs is None:
423 previously_seen_type_refs = set()
424 token = _generic_recursion_cache.set(previously_seen_type_refs)
425 else:
426 token = None
428 try:
429 type_ref = _type_refs.model_type_ref(origin, args_override=args)
430 if type_ref in previously_seen_type_refs:
431 self_type = PydanticRecursiveRef(type_ref=type_ref)
432 yield self_type
433 else:
434 previously_seen_type_refs.add(type_ref)
435 yield
436 previously_seen_type_refs.remove(type_ref)
437 finally:
438 if token:
439 _generic_recursion_cache.reset(token)
442def recursively_defined_type_refs() -> set[str]:
443 visited = _generic_recursion_cache.get()
444 if not visited:
445 return set() # not in a generic recursion, so there are no types
447 return visited.copy() # don't allow modifications
450def _generic_cache_get(key: GenericTypesCacheKey) -> type[BaseModel] | None:
451 try:
452 return _GENERIC_TYPES_CACHE.get(key)
453 except TypeError: # unhashable typevar values
454 return None
457def _generic_cache_set(key: GenericTypesCacheKey, value: type[BaseModel]) -> None:
458 try:
459 _GENERIC_TYPES_CACHE[key] = value
460 except TypeError: # unhashable typevar values
461 pass
464def get_cached_generic_type_early(parent: type[BaseModel], typevar_values: Any) -> type[BaseModel] | None:
465 """The use of a two-stage cache lookup approach was necessary to have the highest performance possible for
466 repeated calls to `__class_getitem__` on generic types (which may happen in tighter loops during runtime),
467 while still ensuring that certain alternative parametrizations ultimately resolve to the same type.
469 As a concrete example, this approach was necessary to make Model[List[T]][int] equal to Model[List[int]].
470 The approach could be modified to not use two different cache keys at different points, but the
471 _early_cache_key is optimized to be as quick to compute as possible (for repeated-access speed), and the
472 _late_cache_key is optimized to be as "correct" as possible, so that two types that will ultimately be the
473 same after resolving the type arguments will always produce cache hits.
475 If we wanted to move to only using a single cache key per type, we would either need to always use the
476 slower/more computationally intensive logic associated with _late_cache_key, or would need to accept
477 that Model[List[T]][int] is a different type than Model[List[T]][int]. Because we rely on subclass relationships
478 during validation, I think it is worthwhile to ensure that types that are functionally equivalent are actually
479 equal.
480 """
481 return _generic_cache_get(_early_cache_key(parent, typevar_values))
484def get_cached_generic_type_late(
485 parent: type[BaseModel], typevar_values: Any, origin: type[BaseModel], args: tuple[Any, ...]
486) -> type[BaseModel] | None:
487 """See the docstring of `get_cached_generic_type_early` for more information about the two-stage cache lookup."""
488 cached = _generic_cache_get(_late_cache_key(origin, args, typevar_values))
489 if cached is not None:
490 set_cached_generic_type(parent, typevar_values, cached, origin, args)
491 return cached
494def set_cached_generic_type(
495 parent: type[BaseModel],
496 typevar_values: tuple[Any, ...],
497 type_: type[BaseModel],
498 origin: type[BaseModel] | None = None,
499 args: tuple[Any, ...] | None = None,
500) -> None:
501 """See the docstring of `get_cached_generic_type_early` for more information about why items are cached with
502 two different keys.
503 """
504 _generic_cache_set(_early_cache_key(parent, typevar_values), type_)
505 if len(typevar_values) == 1:
506 _generic_cache_set(_early_cache_key(parent, typevar_values[0]), type_)
507 if origin and args:
508 _generic_cache_set(_late_cache_key(origin, args, typevar_values), type_)
511def _union_orderings_key(typevar_values: Any) -> Any:
512 """This is intended to help differentiate between Union types with the same arguments in different order.
514 Thanks to caching internal to the `typing` module, it is not possible to distinguish between
515 List[Union[int, float]] and List[Union[float, int]] (and similarly for other "parent" origins besides List)
516 because `typing` considers Union[int, float] to be equal to Union[float, int].
518 However, you _can_ distinguish between (top-level) Union[int, float] vs. Union[float, int].
519 Because we parse items as the first Union type that is successful, we get slightly more consistent behavior
520 if we make an effort to distinguish the ordering of items in a union. It would be best if we could _always_
521 get the exact-correct order of items in the union, but that would require a change to the `typing` module itself.
522 (See https://github.com/python/cpython/issues/86483 for reference.)
523 """
524 if isinstance(typevar_values, tuple):
525 return tuple(_union_orderings_key(value) for value in typevar_values)
526 elif typing_objects.is_union(typing_extensions.get_origin(typevar_values)):
527 return get_args(typevar_values)
528 else:
529 return ()
532def _early_cache_key(cls: type[BaseModel], typevar_values: Any) -> GenericTypesCacheKey:
533 """This is intended for minimal computational overhead during lookups of cached types.
535 Note that this is overly simplistic, and it's possible that two different cls/typevar_values
536 inputs would ultimately result in the same type being created in BaseModel.__class_getitem__.
537 To handle this, we have a fallback _late_cache_key that is checked later if the _early_cache_key
538 lookup fails, and should result in a cache hit _precisely_ when the inputs to __class_getitem__
539 would result in the same type.
540 """
541 return cls, typevar_values, _union_orderings_key(typevar_values)
544def _late_cache_key(origin: type[BaseModel], args: tuple[Any, ...], typevar_values: Any) -> GenericTypesCacheKey:
545 """This is intended for use later in the process of creating a new type, when we have more information
546 about the exact args that will be passed. If it turns out that a different set of inputs to
547 __class_getitem__ resulted in the same inputs to the generic type creation process, we can still
548 return the cached type, and update the cache with the _early_cache_key as well.
549 """
550 # The _union_orderings_key is placed at the start here to ensure there cannot be a collision with an
551 # _early_cache_key, as that function will always produce a BaseModel subclass as the first item in the key,
552 # whereas this function will always produce a tuple as the first item in the key.
553 return _union_orderings_key(typevar_values), origin, args