models.py 7.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193
  1. import inspect
  2. import sys
  3. from collections.abc import Callable
  4. from dataclasses import dataclass, field
  5. from functools import cached_property, partial
  6. from typing import Any, Literal
  7. from fastapi._compat import ModelField
  8. from fastapi.security.base import SecurityBase
  9. from fastapi.types import DependencyCacheKey
  10. if sys.version_info >= (3, 13): # pragma: no cover
  11. from inspect import iscoroutinefunction
  12. else: # pragma: no cover
  13. from asyncio import iscoroutinefunction
  14. def _unwrapped_call(call: Callable[..., Any] | None) -> Any:
  15. if call is None:
  16. return call # pragma: no cover
  17. unwrapped = inspect.unwrap(_impartial(call))
  18. return unwrapped
  19. def _impartial(func: Callable[..., Any]) -> Callable[..., Any]:
  20. while isinstance(func, partial):
  21. func = func.func
  22. return func
  23. @dataclass
  24. class Dependant:
  25. path_params: list[ModelField] = field(default_factory=list)
  26. query_params: list[ModelField] = field(default_factory=list)
  27. header_params: list[ModelField] = field(default_factory=list)
  28. cookie_params: list[ModelField] = field(default_factory=list)
  29. body_params: list[ModelField] = field(default_factory=list)
  30. dependencies: list["Dependant"] = field(default_factory=list)
  31. name: str | None = None
  32. call: Callable[..., Any] | None = None
  33. request_param_name: str | None = None
  34. websocket_param_name: str | None = None
  35. http_connection_param_name: str | None = None
  36. response_param_name: str | None = None
  37. background_tasks_param_name: str | None = None
  38. security_scopes_param_name: str | None = None
  39. own_oauth_scopes: list[str] | None = None
  40. parent_oauth_scopes: list[str] | None = None
  41. use_cache: bool = True
  42. path: str | None = None
  43. scope: Literal["function", "request"] | None = None
  44. @cached_property
  45. def oauth_scopes(self) -> list[str]:
  46. scopes = self.parent_oauth_scopes.copy() if self.parent_oauth_scopes else []
  47. # This doesn't use a set to preserve order, just in case
  48. for scope in self.own_oauth_scopes or []:
  49. if scope not in scopes:
  50. scopes.append(scope)
  51. return scopes
  52. @cached_property
  53. def cache_key(self) -> DependencyCacheKey:
  54. scopes_for_cache = (
  55. tuple(sorted(set(self.oauth_scopes or []))) if self._uses_scopes else ()
  56. )
  57. return (
  58. self.call,
  59. scopes_for_cache,
  60. self.computed_scope or "",
  61. )
  62. @cached_property
  63. def _uses_scopes(self) -> bool:
  64. if self.own_oauth_scopes:
  65. return True
  66. if self.security_scopes_param_name is not None:
  67. return True
  68. if self._is_security_scheme:
  69. return True
  70. for sub_dep in self.dependencies:
  71. if sub_dep._uses_scopes:
  72. return True
  73. return False
  74. @cached_property
  75. def _is_security_scheme(self) -> bool:
  76. if self.call is None:
  77. return False # pragma: no cover
  78. unwrapped = _unwrapped_call(self.call)
  79. return isinstance(unwrapped, SecurityBase)
  80. # Mainly to get the type of SecurityBase, but it's the same self.call
  81. @cached_property
  82. def _security_scheme(self) -> SecurityBase:
  83. unwrapped = _unwrapped_call(self.call)
  84. assert isinstance(unwrapped, SecurityBase)
  85. return unwrapped
  86. @cached_property
  87. def _security_dependencies(self) -> list["Dependant"]:
  88. security_deps = [dep for dep in self.dependencies if dep._is_security_scheme]
  89. return security_deps
  90. @cached_property
  91. def is_gen_callable(self) -> bool:
  92. if self.call is None:
  93. return False # pragma: no cover
  94. if inspect.isgeneratorfunction(
  95. _impartial(self.call)
  96. ) or inspect.isgeneratorfunction(_unwrapped_call(self.call)):
  97. return True
  98. if inspect.isclass(_unwrapped_call(self.call)):
  99. return False
  100. dunder_call = getattr(_impartial(self.call), "__call__", None) # noqa: B004
  101. if dunder_call is None:
  102. return False # pragma: no cover
  103. if inspect.isgeneratorfunction(
  104. _impartial(dunder_call)
  105. ) or inspect.isgeneratorfunction(_unwrapped_call(dunder_call)):
  106. return True
  107. dunder_unwrapped_call = getattr(_unwrapped_call(self.call), "__call__", None) # noqa: B004
  108. if dunder_unwrapped_call is None:
  109. return False # pragma: no cover
  110. if inspect.isgeneratorfunction(
  111. _impartial(dunder_unwrapped_call)
  112. ) or inspect.isgeneratorfunction(_unwrapped_call(dunder_unwrapped_call)):
  113. return True
  114. return False
  115. @cached_property
  116. def is_async_gen_callable(self) -> bool:
  117. if self.call is None:
  118. return False # pragma: no cover
  119. if inspect.isasyncgenfunction(
  120. _impartial(self.call)
  121. ) or inspect.isasyncgenfunction(_unwrapped_call(self.call)):
  122. return True
  123. if inspect.isclass(_unwrapped_call(self.call)):
  124. return False
  125. dunder_call = getattr(_impartial(self.call), "__call__", None) # noqa: B004
  126. if dunder_call is None:
  127. return False # pragma: no cover
  128. if inspect.isasyncgenfunction(
  129. _impartial(dunder_call)
  130. ) or inspect.isasyncgenfunction(_unwrapped_call(dunder_call)):
  131. return True
  132. dunder_unwrapped_call = getattr(_unwrapped_call(self.call), "__call__", None) # noqa: B004
  133. if dunder_unwrapped_call is None:
  134. return False # pragma: no cover
  135. if inspect.isasyncgenfunction(
  136. _impartial(dunder_unwrapped_call)
  137. ) or inspect.isasyncgenfunction(_unwrapped_call(dunder_unwrapped_call)):
  138. return True
  139. return False
  140. @cached_property
  141. def is_coroutine_callable(self) -> bool:
  142. if self.call is None:
  143. return False # pragma: no cover
  144. if inspect.isroutine(_impartial(self.call)) and iscoroutinefunction(
  145. _impartial(self.call)
  146. ):
  147. return True
  148. if inspect.isroutine(_unwrapped_call(self.call)) and iscoroutinefunction(
  149. _unwrapped_call(self.call)
  150. ):
  151. return True
  152. if inspect.isclass(_unwrapped_call(self.call)):
  153. return False
  154. dunder_call = getattr(_impartial(self.call), "__call__", None) # noqa: B004
  155. if dunder_call is None:
  156. return False # pragma: no cover
  157. if iscoroutinefunction(_impartial(dunder_call)) or iscoroutinefunction(
  158. _unwrapped_call(dunder_call)
  159. ):
  160. return True
  161. dunder_unwrapped_call = getattr(_unwrapped_call(self.call), "__call__", None) # noqa: B004
  162. if dunder_unwrapped_call is None:
  163. return False # pragma: no cover
  164. if iscoroutinefunction(
  165. _impartial(dunder_unwrapped_call)
  166. ) or iscoroutinefunction(_unwrapped_call(dunder_unwrapped_call)):
  167. return True
  168. return False
  169. @cached_property
  170. def computed_scope(self) -> str | None:
  171. if self.scope:
  172. return self.scope
  173. if self.is_gen_callable or self.is_async_gen_callable:
  174. return "request"
  175. return None