shared.py 6.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214
  1. import types
  2. import typing
  3. import warnings
  4. from collections import deque
  5. from collections.abc import Mapping, Sequence
  6. from dataclasses import is_dataclass
  7. from typing import (
  8. Annotated,
  9. Any,
  10. TypeGuard,
  11. TypeVar,
  12. Union,
  13. get_args,
  14. get_origin,
  15. )
  16. from fastapi.types import UnionType
  17. from pydantic import BaseModel
  18. from pydantic.version import VERSION as PYDANTIC_VERSION
  19. from starlette.datastructures import UploadFile
  20. _T = TypeVar("_T")
  21. # Copy from Pydantic: pydantic/_internal/_typing_extra.py
  22. WithArgsTypes: tuple[Any, ...] = (
  23. typing._GenericAlias, # type: ignore[attr-defined]
  24. types.GenericAlias,
  25. types.UnionType,
  26. ) # pyright: ignore[reportAttributeAccessIssue]
  27. PYDANTIC_VERSION_MINOR_TUPLE = tuple(int(x) for x in PYDANTIC_VERSION.split(".")[:2])
  28. sequence_annotation_to_type = {
  29. Sequence: list,
  30. list: list,
  31. tuple: tuple,
  32. set: set,
  33. frozenset: frozenset,
  34. deque: deque,
  35. }
  36. sequence_types: tuple[type[Any], ...] = tuple(sequence_annotation_to_type.keys())
  37. # Copy of Pydantic: pydantic/_internal/_utils.py with added TypeGuard
  38. def lenient_issubclass(
  39. cls: Any, class_or_tuple: type[_T] | tuple[type[_T], ...] | None
  40. ) -> TypeGuard[type[_T]]:
  41. try:
  42. return isinstance(cls, type) and issubclass(cls, class_or_tuple) # type: ignore[arg-type]
  43. except TypeError: # pragma: no cover
  44. if isinstance(cls, WithArgsTypes):
  45. return False
  46. raise # pragma: no cover
  47. def _annotation_is_sequence(annotation: type[Any] | None) -> bool:
  48. if lenient_issubclass(annotation, (str, bytes)):
  49. return False
  50. return lenient_issubclass(annotation, sequence_types)
  51. def field_annotation_is_sequence(annotation: type[Any] | None) -> bool:
  52. origin = get_origin(annotation)
  53. if origin is Union or origin is UnionType:
  54. for arg in get_args(annotation):
  55. if field_annotation_is_sequence(arg):
  56. return True
  57. return False
  58. return _annotation_is_sequence(annotation) or _annotation_is_sequence(
  59. get_origin(annotation)
  60. )
  61. def value_is_sequence(value: Any) -> bool:
  62. return isinstance(value, sequence_types) and not isinstance(value, (str, bytes))
  63. def _annotation_is_complex(annotation: type[Any] | None) -> bool:
  64. return (
  65. lenient_issubclass(annotation, (BaseModel, Mapping, UploadFile))
  66. or _annotation_is_sequence(annotation)
  67. or is_dataclass(annotation)
  68. )
  69. def field_annotation_is_complex(annotation: type[Any] | None) -> bool:
  70. origin = get_origin(annotation)
  71. if origin is Union or origin is UnionType:
  72. return any(field_annotation_is_complex(arg) for arg in get_args(annotation))
  73. if origin is Annotated:
  74. return field_annotation_is_complex(get_args(annotation)[0])
  75. return (
  76. _annotation_is_complex(annotation)
  77. or _annotation_is_complex(origin)
  78. or hasattr(origin, "__pydantic_core_schema__")
  79. or hasattr(origin, "__get_pydantic_core_schema__")
  80. )
  81. def field_annotation_is_scalar(annotation: Any) -> bool:
  82. # handle Ellipsis here to make tuple[int, ...] work nicely
  83. return annotation is Ellipsis or not field_annotation_is_complex(annotation)
  84. def field_annotation_is_scalar_sequence(annotation: type[Any] | None) -> bool:
  85. origin = get_origin(annotation)
  86. if origin is Union or origin is UnionType:
  87. at_least_one_scalar_sequence = False
  88. for arg in get_args(annotation):
  89. if field_annotation_is_scalar_sequence(arg):
  90. at_least_one_scalar_sequence = True
  91. continue
  92. elif not field_annotation_is_scalar(arg):
  93. return False
  94. return at_least_one_scalar_sequence
  95. return field_annotation_is_sequence(annotation) and all(
  96. field_annotation_is_scalar(sub_annotation)
  97. for sub_annotation in get_args(annotation)
  98. )
  99. def is_bytes_or_nonable_bytes_annotation(annotation: Any) -> bool:
  100. if lenient_issubclass(annotation, bytes):
  101. return True
  102. origin = get_origin(annotation)
  103. if origin is Union or origin is UnionType:
  104. for arg in get_args(annotation):
  105. if lenient_issubclass(arg, bytes):
  106. return True
  107. return False
  108. def is_uploadfile_or_nonable_uploadfile_annotation(annotation: Any) -> bool:
  109. if lenient_issubclass(annotation, UploadFile):
  110. return True
  111. origin = get_origin(annotation)
  112. if origin is Union or origin is UnionType:
  113. for arg in get_args(annotation):
  114. if lenient_issubclass(arg, UploadFile):
  115. return True
  116. return False
  117. def is_bytes_sequence_annotation(annotation: Any) -> bool:
  118. origin = get_origin(annotation)
  119. if origin is Union or origin is UnionType:
  120. at_least_one = False
  121. for arg in get_args(annotation):
  122. if is_bytes_sequence_annotation(arg):
  123. at_least_one = True
  124. continue
  125. return at_least_one
  126. return field_annotation_is_sequence(annotation) and all(
  127. is_bytes_or_nonable_bytes_annotation(sub_annotation)
  128. for sub_annotation in get_args(annotation)
  129. )
  130. def is_uploadfile_sequence_annotation(annotation: Any) -> bool:
  131. origin = get_origin(annotation)
  132. if origin is Union or origin is UnionType:
  133. at_least_one = False
  134. for arg in get_args(annotation):
  135. if is_uploadfile_sequence_annotation(arg):
  136. at_least_one = True
  137. continue
  138. return at_least_one
  139. return field_annotation_is_sequence(annotation) and all(
  140. is_uploadfile_or_nonable_uploadfile_annotation(sub_annotation)
  141. for sub_annotation in get_args(annotation)
  142. )
  143. def is_pydantic_v1_model_instance(obj: Any) -> bool:
  144. # TODO: remove this function once the required version of Pydantic fully
  145. # removes pydantic.v1
  146. try:
  147. with warnings.catch_warnings():
  148. warnings.simplefilter("ignore", UserWarning)
  149. from pydantic import v1
  150. except ImportError: # pragma: no cover
  151. return False
  152. return isinstance(obj, v1.BaseModel)
  153. def is_pydantic_v1_model_class(cls: Any) -> bool:
  154. # TODO: remove this function once the required version of Pydantic fully
  155. # removes pydantic.v1
  156. try:
  157. with warnings.catch_warnings():
  158. warnings.simplefilter("ignore", UserWarning)
  159. from pydantic import v1
  160. except ImportError: # pragma: no cover
  161. return False
  162. return lenient_issubclass(cls, v1.BaseModel)
  163. def annotation_is_pydantic_v1(annotation: Any) -> bool:
  164. if is_pydantic_v1_model_class(annotation):
  165. return True
  166. origin = get_origin(annotation)
  167. if origin is Union or origin is UnionType:
  168. for arg in get_args(annotation):
  169. if is_pydantic_v1_model_class(arg):
  170. return True
  171. if field_annotation_is_sequence(annotation):
  172. for sub_annotation in get_args(annotation):
  173. if annotation_is_pydantic_v1(sub_annotation):
  174. return True
  175. return False