_utils.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335
  1. """Bucket of reusable internal utilities.
  2. This should be reduced as much as possible with functions only used in one place, moved to that place.
  3. """
  4. from __future__ import annotations as _annotations
  5. import keyword
  6. import typing
  7. import weakref
  8. from collections import OrderedDict, defaultdict, deque
  9. from copy import deepcopy
  10. from itertools import zip_longest
  11. from types import BuiltinFunctionType, CodeType, FunctionType, GeneratorType, LambdaType, ModuleType
  12. from typing import Any, TypeVar
  13. from typing_extensions import TypeAlias, TypeGuard
  14. from . import _repr, _typing_extra
  15. if typing.TYPE_CHECKING:
  16. MappingIntStrAny: TypeAlias = 'typing.Mapping[int, Any] | typing.Mapping[str, Any]'
  17. AbstractSetIntStr: TypeAlias = 'typing.AbstractSet[int] | typing.AbstractSet[str]'
  18. from ..main import BaseModel
  19. # these are types that are returned unchanged by deepcopy
  20. IMMUTABLE_NON_COLLECTIONS_TYPES: set[type[Any]] = {
  21. int,
  22. float,
  23. complex,
  24. str,
  25. bool,
  26. bytes,
  27. type,
  28. _typing_extra.NoneType,
  29. FunctionType,
  30. BuiltinFunctionType,
  31. LambdaType,
  32. weakref.ref,
  33. CodeType,
  34. # note: including ModuleType will differ from behaviour of deepcopy by not producing error.
  35. # It might be not a good idea in general, but considering that this function used only internally
  36. # against default values of fields, this will allow to actually have a field with module as default value
  37. ModuleType,
  38. NotImplemented.__class__,
  39. Ellipsis.__class__,
  40. }
  41. # these are types that if empty, might be copied with simple copy() instead of deepcopy()
  42. BUILTIN_COLLECTIONS: set[type[Any]] = {
  43. list,
  44. set,
  45. tuple,
  46. frozenset,
  47. dict,
  48. OrderedDict,
  49. defaultdict,
  50. deque,
  51. }
  52. def sequence_like(v: Any) -> bool:
  53. return isinstance(v, (list, tuple, set, frozenset, GeneratorType, deque))
  54. def lenient_isinstance(o: Any, class_or_tuple: type[Any] | tuple[type[Any], ...] | None) -> bool: # pragma: no cover
  55. try:
  56. return isinstance(o, class_or_tuple) # type: ignore[arg-type]
  57. except TypeError:
  58. return False
  59. def lenient_issubclass(cls: Any, class_or_tuple: Any) -> bool: # pragma: no cover
  60. try:
  61. return isinstance(cls, type) and issubclass(cls, class_or_tuple)
  62. except TypeError:
  63. if isinstance(cls, _typing_extra.WithArgsTypes):
  64. return False
  65. raise # pragma: no cover
  66. def is_model_class(cls: Any) -> TypeGuard[type[BaseModel]]:
  67. """Returns true if cls is a _proper_ subclass of BaseModel, and provides proper type-checking,
  68. unlike raw calls to lenient_issubclass.
  69. """
  70. from ..main import BaseModel
  71. return lenient_issubclass(cls, BaseModel) and cls is not BaseModel
  72. def is_valid_identifier(identifier: str) -> bool:
  73. """Checks that a string is a valid identifier and not a Python keyword.
  74. :param identifier: The identifier to test.
  75. :return: True if the identifier is valid.
  76. """
  77. return identifier.isidentifier() and not keyword.iskeyword(identifier)
  78. KeyType = TypeVar('KeyType')
  79. def deep_update(mapping: dict[KeyType, Any], *updating_mappings: dict[KeyType, Any]) -> dict[KeyType, Any]:
  80. updated_mapping = mapping.copy()
  81. for updating_mapping in updating_mappings:
  82. for k, v in updating_mapping.items():
  83. if k in updated_mapping and isinstance(updated_mapping[k], dict) and isinstance(v, dict):
  84. updated_mapping[k] = deep_update(updated_mapping[k], v)
  85. else:
  86. updated_mapping[k] = v
  87. return updated_mapping
  88. def update_not_none(mapping: dict[Any, Any], **update: Any) -> None:
  89. mapping.update({k: v for k, v in update.items() if v is not None})
  90. T = TypeVar('T')
  91. def unique_list(
  92. input_list: list[T] | tuple[T, ...],
  93. *,
  94. name_factory: typing.Callable[[T], str] = str,
  95. ) -> list[T]:
  96. """Make a list unique while maintaining order.
  97. We update the list if another one with the same name is set
  98. (e.g. model validator overridden in subclass).
  99. """
  100. result: list[T] = []
  101. result_names: list[str] = []
  102. for v in input_list:
  103. v_name = name_factory(v)
  104. if v_name not in result_names:
  105. result_names.append(v_name)
  106. result.append(v)
  107. else:
  108. result[result_names.index(v_name)] = v
  109. return result
  110. class ValueItems(_repr.Representation):
  111. """Class for more convenient calculation of excluded or included fields on values."""
  112. __slots__ = ('_items', '_type')
  113. def __init__(self, value: Any, items: AbstractSetIntStr | MappingIntStrAny) -> None:
  114. items = self._coerce_items(items)
  115. if isinstance(value, (list, tuple)):
  116. items = self._normalize_indexes(items, len(value)) # type: ignore
  117. self._items: MappingIntStrAny = items # type: ignore
  118. def is_excluded(self, item: Any) -> bool:
  119. """Check if item is fully excluded.
  120. :param item: key or index of a value
  121. """
  122. return self.is_true(self._items.get(item))
  123. def is_included(self, item: Any) -> bool:
  124. """Check if value is contained in self._items.
  125. :param item: key or index of value
  126. """
  127. return item in self._items
  128. def for_element(self, e: int | str) -> AbstractSetIntStr | MappingIntStrAny | None:
  129. """:param e: key or index of element on value
  130. :return: raw values for element if self._items is dict and contain needed element
  131. """
  132. item = self._items.get(e) # type: ignore
  133. return item if not self.is_true(item) else None
  134. def _normalize_indexes(self, items: MappingIntStrAny, v_length: int) -> dict[int | str, Any]:
  135. """:param items: dict or set of indexes which will be normalized
  136. :param v_length: length of sequence indexes of which will be
  137. >>> self._normalize_indexes({0: True, -2: True, -1: True}, 4)
  138. {0: True, 2: True, 3: True}
  139. >>> self._normalize_indexes({'__all__': True}, 4)
  140. {0: True, 1: True, 2: True, 3: True}
  141. """
  142. normalized_items: dict[int | str, Any] = {}
  143. all_items = None
  144. for i, v in items.items():
  145. if not (isinstance(v, typing.Mapping) or isinstance(v, typing.AbstractSet) or self.is_true(v)):
  146. raise TypeError(f'Unexpected type of exclude value for index "{i}" {v.__class__}')
  147. if i == '__all__':
  148. all_items = self._coerce_value(v)
  149. continue
  150. if not isinstance(i, int):
  151. raise TypeError(
  152. 'Excluding fields from a sequence of sub-models or dicts must be performed index-wise: '
  153. 'expected integer keys or keyword "__all__"'
  154. )
  155. normalized_i = v_length + i if i < 0 else i
  156. normalized_items[normalized_i] = self.merge(v, normalized_items.get(normalized_i))
  157. if not all_items:
  158. return normalized_items
  159. if self.is_true(all_items):
  160. for i in range(v_length):
  161. normalized_items.setdefault(i, ...)
  162. return normalized_items
  163. for i in range(v_length):
  164. normalized_item = normalized_items.setdefault(i, {})
  165. if not self.is_true(normalized_item):
  166. normalized_items[i] = self.merge(all_items, normalized_item)
  167. return normalized_items
  168. @classmethod
  169. def merge(cls, base: Any, override: Any, intersect: bool = False) -> Any:
  170. """Merge a `base` item with an `override` item.
  171. Both `base` and `override` are converted to dictionaries if possible.
  172. Sets are converted to dictionaries with the sets entries as keys and
  173. Ellipsis as values.
  174. Each key-value pair existing in `base` is merged with `override`,
  175. while the rest of the key-value pairs are updated recursively with this function.
  176. Merging takes place based on the "union" of keys if `intersect` is
  177. set to `False` (default) and on the intersection of keys if
  178. `intersect` is set to `True`.
  179. """
  180. override = cls._coerce_value(override)
  181. base = cls._coerce_value(base)
  182. if override is None:
  183. return base
  184. if cls.is_true(base) or base is None:
  185. return override
  186. if cls.is_true(override):
  187. return base if intersect else override
  188. # intersection or union of keys while preserving ordering:
  189. if intersect:
  190. merge_keys = [k for k in base if k in override] + [k for k in override if k in base]
  191. else:
  192. merge_keys = list(base) + [k for k in override if k not in base]
  193. merged: dict[int | str, Any] = {}
  194. for k in merge_keys:
  195. merged_item = cls.merge(base.get(k), override.get(k), intersect=intersect)
  196. if merged_item is not None:
  197. merged[k] = merged_item
  198. return merged
  199. @staticmethod
  200. def _coerce_items(items: AbstractSetIntStr | MappingIntStrAny) -> MappingIntStrAny:
  201. if isinstance(items, typing.Mapping):
  202. pass
  203. elif isinstance(items, typing.AbstractSet):
  204. items = dict.fromkeys(items, ...) # type: ignore
  205. else:
  206. class_name = getattr(items, '__class__', '???')
  207. raise TypeError(f'Unexpected type of exclude value {class_name}')
  208. return items # type: ignore
  209. @classmethod
  210. def _coerce_value(cls, value: Any) -> Any:
  211. if value is None or cls.is_true(value):
  212. return value
  213. return cls._coerce_items(value)
  214. @staticmethod
  215. def is_true(v: Any) -> bool:
  216. return v is True or v is ...
  217. def __repr_args__(self) -> _repr.ReprArgs:
  218. return [(None, self._items)]
  219. if typing.TYPE_CHECKING:
  220. def ClassAttribute(name: str, value: T) -> T:
  221. ...
  222. else:
  223. class ClassAttribute:
  224. """Hide class attribute from its instances."""
  225. __slots__ = 'name', 'value'
  226. def __init__(self, name: str, value: Any) -> None:
  227. self.name = name
  228. self.value = value
  229. def __get__(self, instance: Any, owner: type[Any]) -> None:
  230. if instance is None:
  231. return self.value
  232. raise AttributeError(f'{self.name!r} attribute of {owner.__name__!r} is class-only')
  233. Obj = TypeVar('Obj')
  234. def smart_deepcopy(obj: Obj) -> Obj:
  235. """Return type as is for immutable built-in types
  236. Use obj.copy() for built-in empty collections
  237. Use copy.deepcopy() for non-empty collections and unknown objects.
  238. """
  239. obj_type = obj.__class__
  240. if obj_type in IMMUTABLE_NON_COLLECTIONS_TYPES:
  241. return obj # fastest case: obj is immutable and not collection therefore will not be copied anyway
  242. try:
  243. if not obj and obj_type in BUILTIN_COLLECTIONS:
  244. # faster way for empty collections, no need to copy its members
  245. return obj if obj_type is tuple else obj.copy() # tuple doesn't have copy method # type: ignore
  246. except (TypeError, ValueError, RuntimeError):
  247. # do we really dare to catch ALL errors? Seems a bit risky
  248. pass
  249. return deepcopy(obj) # slowest way when we actually might need a deepcopy
  250. _EMPTY = object()
  251. def all_identical(left: typing.Iterable[Any], right: typing.Iterable[Any]) -> bool:
  252. """Check that the items of `left` are the same objects as those in `right`.
  253. >>> a, b = object(), object()
  254. >>> all_identical([a, b, a], [a, b, a])
  255. True
  256. >>> all_identical([a, b, [a]], [a, b, [a]]) # new list object, while "equal" is not "identical"
  257. False
  258. """
  259. for left_item, right_item in zip_longest(left, right, fillvalue=_EMPTY):
  260. if left_item is not right_item:
  261. return False
  262. return True