Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""Bucket of reusable internal utilities.
3This should be reduced as much as possible with functions only used in one place, moved to that place.
4"""
6from __future__ import annotations as _annotations
8import dataclasses
9import keyword
10import warnings
11import weakref
12from collections import OrderedDict, defaultdict, deque
13from collections.abc import Callable, Iterable, Mapping
14from collections.abc import Set as AbstractSet
15from copy import deepcopy
16from functools import cached_property
17from inspect import Parameter
18from itertools import zip_longest
19from types import BuiltinFunctionType, CodeType, FunctionType, GeneratorType, LambdaType, ModuleType, NoneType
20from typing import TYPE_CHECKING, Any, Generic, TypeAlias, TypeGuard, TypeVar, overload
22from pydantic_core import MISSING, PydanticUndefined
23from typing_extensions import deprecated
25from pydantic import PydanticDeprecatedSince211
27from . import _repr, _typing_extra
28from ._import_utils import import_cached_base_model
30if TYPE_CHECKING:
31 MappingIntStrAny: TypeAlias = Mapping[int, Any] | Mapping[str, Any]
32 AbstractSetIntStr: TypeAlias = AbstractSet[int] | AbstractSet[str]
33 from ..main import BaseModel
36# these are types that are returned unchanged by deepcopy
37IMMUTABLE_NON_COLLECTIONS_TYPES: set[type[Any]] = {
38 int,
39 float,
40 complex,
41 str,
42 bool,
43 bytes,
44 type,
45 NoneType,
46 FunctionType,
47 BuiltinFunctionType,
48 LambdaType,
49 weakref.ref,
50 CodeType,
51 # note: including ModuleType will differ from behaviour of deepcopy by not producing error.
52 # It might be not a good idea in general, but considering that this function used only internally
53 # against default values of fields, this will allow one to actually have a field with module as default value
54 ModuleType,
55 NotImplemented.__class__,
56 Ellipsis.__class__,
57}
59# these are types that if empty, might be copied with simple copy() instead of deepcopy()
60BUILTIN_COLLECTIONS: set[type[Any]] = {
61 list,
62 set,
63 tuple,
64 frozenset,
65 dict,
66 OrderedDict,
67 defaultdict,
68 deque,
69}
72def can_be_positional(param: Parameter) -> bool:
73 """Return whether the parameter accepts a positional argument.
75 ```python {test="skip" lint="skip"}
76 def func(a, /, b, *, c):
77 pass
79 params = inspect.signature(func).parameters
80 can_be_positional(params['a'])
81 #> True
82 can_be_positional(params['b'])
83 #> True
84 can_be_positional(params['c'])
85 #> False
86 ```
87 """
88 return param.kind in (Parameter.POSITIONAL_ONLY, Parameter.POSITIONAL_OR_KEYWORD)
91def sequence_like(v: Any) -> bool:
92 return isinstance(v, (list, tuple, set, frozenset, GeneratorType, deque))
95def lenient_isinstance(o: Any, class_or_tuple: type[Any] | tuple[type[Any], ...] | None) -> bool: # pragma: no cover
96 try:
97 return isinstance(o, class_or_tuple) # type: ignore[arg-type]
98 except TypeError:
99 return False
102def lenient_issubclass(cls: Any, class_or_tuple: Any) -> bool: # pragma: no cover
103 try:
104 return isinstance(cls, type) and issubclass(cls, class_or_tuple)
105 except TypeError:
106 if isinstance(cls, _typing_extra.WithArgsTypes):
107 return False
108 raise # pragma: no cover
111def is_model_class(cls: Any) -> TypeGuard[type[BaseModel]]:
112 """Returns true if cls is a _proper_ subclass of BaseModel, and provides proper type-checking,
113 unlike raw calls to lenient_issubclass.
114 """
115 BaseModel = import_cached_base_model()
117 return lenient_issubclass(cls, BaseModel) and cls is not BaseModel
120def is_valid_identifier(identifier: str) -> bool:
121 """Checks that a string is a valid identifier and not a Python keyword.
122 :param identifier: The identifier to test.
123 :return: True if the identifier is valid.
124 """
125 return identifier.isidentifier() and not keyword.iskeyword(identifier)
128KeyType = TypeVar('KeyType')
131def deep_update(mapping: dict[KeyType, Any], *updating_mappings: dict[KeyType, Any]) -> dict[KeyType, Any]:
132 updated_mapping = mapping.copy()
133 for updating_mapping in updating_mappings:
134 for k, v in updating_mapping.items():
135 if k in updated_mapping and isinstance(updated_mapping[k], dict) and isinstance(v, dict):
136 updated_mapping[k] = deep_update(updated_mapping[k], v)
137 else:
138 updated_mapping[k] = v
139 return updated_mapping
142def update_not_none(mapping: dict[Any, Any], **update: Any) -> None:
143 mapping.update({k: v for k, v in update.items() if v is not None})
146T = TypeVar('T')
149def unique_list(
150 input_list: list[T] | tuple[T, ...],
151 *,
152 name_factory: Callable[[T], str] = str,
153) -> list[T]:
154 """Make a list unique while maintaining order.
155 We update the list if another one with the same name is set
156 (e.g. model validator overridden in subclass).
157 """
158 result: list[T] = []
159 result_names: list[str] = []
160 for v in input_list:
161 v_name = name_factory(v)
162 if v_name not in result_names:
163 result_names.append(v_name)
164 result.append(v)
165 else:
166 result[result_names.index(v_name)] = v
168 return result
171class ValueItems(_repr.Representation):
172 """Class for more convenient calculation of excluded or included fields on values."""
174 __slots__ = ('_items',)
176 def __init__(self, value: Any, items: AbstractSetIntStr | MappingIntStrAny) -> None:
177 items = self._coerce_items(items)
179 if isinstance(value, (list, tuple)):
180 items = self._normalize_indexes(items, len(value)) # type: ignore
182 self._items: MappingIntStrAny = items # type: ignore
184 def is_excluded(self, item: Any) -> bool:
185 """Check if item is fully excluded.
187 :param item: key or index of a value
188 """
189 return self.is_true(self._items.get(item))
191 def is_included(self, item: Any) -> bool:
192 """Check if value is contained in self._items.
194 :param item: key or index of value
195 """
196 return item in self._items
198 def for_element(self, e: int | str) -> AbstractSetIntStr | MappingIntStrAny | None:
199 """:param e: key or index of element on value
200 :return: raw values for element if self._items is dict and contain needed element
201 """
202 item = self._items.get(e) # type: ignore
203 return item if not self.is_true(item) else None
205 def _normalize_indexes(self, items: MappingIntStrAny, v_length: int) -> dict[int | str, Any]:
206 """:param items: dict or set of indexes which will be normalized
207 :param v_length: length of sequence indexes of which will be
209 >>> self._normalize_indexes({0: True, -2: True, -1: True}, 4)
210 {0: True, 2: True, 3: True}
211 >>> self._normalize_indexes({'__all__': True}, 4)
212 {0: True, 1: True, 2: True, 3: True}
213 """
214 normalized_items: dict[int | str, Any] = {}
215 all_items = None
216 for i, v in items.items():
217 if not (isinstance(v, Mapping) or isinstance(v, AbstractSet) or self.is_true(v)):
218 raise TypeError(f'Unexpected type of exclude value for index "{i}" {v.__class__}')
219 if i == '__all__':
220 all_items = self._coerce_value(v)
221 continue
222 if not isinstance(i, int):
223 raise TypeError(
224 'Excluding fields from a sequence of sub-models or dicts must be performed index-wise: '
225 'expected integer keys or keyword "__all__"'
226 )
227 normalized_i = v_length + i if i < 0 else i
228 normalized_items[normalized_i] = self.merge(v, normalized_items.get(normalized_i))
230 if not all_items:
231 return normalized_items
232 if self.is_true(all_items):
233 for i in range(v_length):
234 normalized_items.setdefault(i, ...)
235 return normalized_items
236 for i in range(v_length):
237 normalized_item = normalized_items.setdefault(i, {})
238 if not self.is_true(normalized_item):
239 normalized_items[i] = self.merge(all_items, normalized_item)
240 return normalized_items
242 @classmethod
243 def merge(cls, base: Any, override: Any, intersect: bool = False) -> Any:
244 """Merge a `base` item with an `override` item.
246 Both `base` and `override` are converted to dictionaries if possible.
247 Sets are converted to dictionaries with the sets entries as keys and
248 Ellipsis as values.
250 Each key-value pair existing in `base` is merged with `override`,
251 while the rest of the key-value pairs are updated recursively with this function.
253 Merging takes place based on the "union" of keys if `intersect` is
254 set to `False` (default) and on the intersection of keys if
255 `intersect` is set to `True`.
256 """
257 override = cls._coerce_value(override)
258 base = cls._coerce_value(base)
259 if override is None:
260 return base
261 if cls.is_true(base) or base is None:
262 return override
263 if cls.is_true(override):
264 return base if intersect else override
266 # intersection or union of keys while preserving ordering:
267 if intersect:
268 merge_keys = [k for k in base if k in override] + [k for k in override if k in base]
269 else:
270 merge_keys = list(base) + [k for k in override if k not in base]
272 merged: dict[int | str, Any] = {}
273 for k in merge_keys:
274 merged_item = cls.merge(base.get(k), override.get(k), intersect=intersect)
275 if merged_item is not None:
276 merged[k] = merged_item
278 return merged
280 @staticmethod
281 def _coerce_items(items: AbstractSetIntStr | MappingIntStrAny) -> MappingIntStrAny:
282 if isinstance(items, Mapping):
283 pass
284 elif isinstance(items, AbstractSet):
285 items = dict.fromkeys(items, ...) # type: ignore
286 else:
287 class_name = getattr(items, '__class__', '???')
288 raise TypeError(f'Unexpected type of exclude value {class_name}')
289 return items # type: ignore
291 @classmethod
292 def _coerce_value(cls, value: Any) -> Any:
293 if value is None or cls.is_true(value):
294 return value
295 return cls._coerce_items(value)
297 @staticmethod
298 def is_true(v: Any) -> bool:
299 return v is True or v is ...
301 def __repr_args__(self) -> _repr.ReprArgs:
302 return [(None, self._items)]
305if TYPE_CHECKING:
307 def LazyClassAttribute(name: str, get_value: Callable[[], T]) -> T: ...
309else:
311 class LazyClassAttribute:
312 """A descriptor exposing an attribute only accessible on a class (hidden from instances).
314 The attribute is lazily computed and cached during the first access.
315 """
317 def __init__(self, name: str, get_value: Callable[[], Any]) -> None:
318 self.name = name
319 self.get_value = get_value
321 @cached_property
322 def value(self) -> Any:
323 return self.get_value()
325 def __get__(self, instance: Any, owner: type[Any]) -> Any:
326 if instance is None:
327 return self.value
328 raise AttributeError(f'{self.name!r} attribute of {owner.__name__!r} is class-only')
331Obj = TypeVar('Obj')
334def smart_deepcopy(obj: Obj) -> Obj:
335 """Return type as is for immutable built-in types
336 Use obj.copy() for built-in empty collections
337 Use copy.deepcopy() for non-empty collections and unknown objects.
338 """
339 if obj is MISSING or obj is PydanticUndefined:
340 return obj # pyright: ignore[reportReturnType]
341 obj_type = obj.__class__
342 if obj_type in IMMUTABLE_NON_COLLECTIONS_TYPES:
343 return obj # fastest case: obj is immutable and not collection therefore will not be copied anyway
344 try:
345 if not obj and obj_type in BUILTIN_COLLECTIONS:
346 # faster way for empty collections, no need to copy its members
347 return obj if obj_type is tuple else obj.copy() # tuple doesn't have copy method # type: ignore
348 except (TypeError, ValueError, RuntimeError):
349 # do we really dare to catch ALL errors? Seems a bit risky
350 pass
352 return deepcopy(obj) # slowest way when we actually might need a deepcopy
355_SENTINEL = object()
358def all_identical(left: Iterable[Any], right: Iterable[Any]) -> bool:
359 """Check that the items of `left` are the same objects as those in `right`.
361 >>> a, b = object(), object()
362 >>> all_identical([a, b, a], [a, b, a])
363 True
364 >>> all_identical([a, b, [a]], [a, b, [a]]) # new list object, while "equal" is not "identical"
365 False
366 """
367 for left_item, right_item in zip_longest(left, right, fillvalue=_SENTINEL):
368 if left_item is not right_item:
369 return False
370 return True
373def get_first_not_none(a: Any, b: Any) -> Any:
374 """Return the first argument if it is not `None`, otherwise return the second argument."""
375 return a if a is not None else b
378@dataclasses.dataclass(frozen=True, slots=True)
379class SafeGetItemProxy:
380 """Wrapper redirecting `__getitem__` to `get` with a sentinel value as default
382 This makes it safe to use in `operator.itemgetter` when some keys may be missing
383 """
385 wrapped: Mapping[str, Any]
387 def __getitem__(self, key: str, /) -> Any:
388 return self.wrapped.get(key, _SENTINEL)
390 # required to pass the object to operator.itemgetter() instances due to a quirk of typeshed
391 # https://github.com/python/mypy/issues/13713
392 # https://github.com/python/typeshed/pull/8785
393 # Since this is typing-only, hide it in a typing.TYPE_CHECKING block
394 if TYPE_CHECKING:
396 def __contains__(self, key: str, /) -> bool:
397 return self.wrapped.__contains__(key)
400_ModelT = TypeVar('_ModelT', bound='BaseModel')
401_RT = TypeVar('_RT')
404class deprecated_instance_property(Generic[_ModelT, _RT]):
405 """A decorator exposing the decorated class method as a property, with a warning on instance access.
407 This decorator takes a class method defined on the `BaseModel` class and transforms it into
408 an attribute. The attribute can be accessed on both the class and instances of the class. If accessed
409 via an instance, a deprecation warning is emitted stating that instance access will be removed in V3.
410 """
412 def __init__(self, fget: Callable[[type[_ModelT]], _RT], /) -> None:
413 # Note: fget should be a classmethod:
414 self.fget = fget
416 @overload
417 def __get__(self, instance: None, objtype: type[_ModelT]) -> _RT: ...
418 @overload
419 @deprecated(
420 'Accessing this attribute on the instance is deprecated, and will be removed in Pydantic V3. '
421 'Instead, you should access this attribute from the model class.',
422 category=None,
423 )
424 def __get__(self, instance: _ModelT, objtype: type[_ModelT]) -> _RT: ...
425 def __get__(self, instance: _ModelT | None, objtype: type[_ModelT]) -> _RT:
426 if instance is not None:
427 warnings.warn(
428 f'Accessing the {self.fget.__name__!r} attribute on the instance is deprecated. '
429 'Instead, you should access this attribute from the model class.',
430 category=PydanticDeprecatedSince211,
431 stacklevel=2,
432 )
433 return self.fget.__get__(instance, objtype)()