| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193 |
- import inspect
- import sys
- from collections.abc import Callable
- from dataclasses import dataclass, field
- from functools import cached_property, partial
- from typing import Any, Literal
- from fastapi._compat import ModelField
- from fastapi.security.base import SecurityBase
- from fastapi.types import DependencyCacheKey
- if sys.version_info >= (3, 13): # pragma: no cover
- from inspect import iscoroutinefunction
- else: # pragma: no cover
- from asyncio import iscoroutinefunction
- def _unwrapped_call(call: Callable[..., Any] | None) -> Any:
- if call is None:
- return call # pragma: no cover
- unwrapped = inspect.unwrap(_impartial(call))
- return unwrapped
- def _impartial(func: Callable[..., Any]) -> Callable[..., Any]:
- while isinstance(func, partial):
- func = func.func
- return func
- @dataclass
- class Dependant:
- path_params: list[ModelField] = field(default_factory=list)
- query_params: list[ModelField] = field(default_factory=list)
- header_params: list[ModelField] = field(default_factory=list)
- cookie_params: list[ModelField] = field(default_factory=list)
- body_params: list[ModelField] = field(default_factory=list)
- dependencies: list["Dependant"] = field(default_factory=list)
- name: str | None = None
- call: Callable[..., Any] | None = None
- request_param_name: str | None = None
- websocket_param_name: str | None = None
- http_connection_param_name: str | None = None
- response_param_name: str | None = None
- background_tasks_param_name: str | None = None
- security_scopes_param_name: str | None = None
- own_oauth_scopes: list[str] | None = None
- parent_oauth_scopes: list[str] | None = None
- use_cache: bool = True
- path: str | None = None
- scope: Literal["function", "request"] | None = None
- @cached_property
- def oauth_scopes(self) -> list[str]:
- scopes = self.parent_oauth_scopes.copy() if self.parent_oauth_scopes else []
- # This doesn't use a set to preserve order, just in case
- for scope in self.own_oauth_scopes or []:
- if scope not in scopes:
- scopes.append(scope)
- return scopes
- @cached_property
- def cache_key(self) -> DependencyCacheKey:
- scopes_for_cache = (
- tuple(sorted(set(self.oauth_scopes or []))) if self._uses_scopes else ()
- )
- return (
- self.call,
- scopes_for_cache,
- self.computed_scope or "",
- )
- @cached_property
- def _uses_scopes(self) -> bool:
- if self.own_oauth_scopes:
- return True
- if self.security_scopes_param_name is not None:
- return True
- if self._is_security_scheme:
- return True
- for sub_dep in self.dependencies:
- if sub_dep._uses_scopes:
- return True
- return False
- @cached_property
- def _is_security_scheme(self) -> bool:
- if self.call is None:
- return False # pragma: no cover
- unwrapped = _unwrapped_call(self.call)
- return isinstance(unwrapped, SecurityBase)
- # Mainly to get the type of SecurityBase, but it's the same self.call
- @cached_property
- def _security_scheme(self) -> SecurityBase:
- unwrapped = _unwrapped_call(self.call)
- assert isinstance(unwrapped, SecurityBase)
- return unwrapped
- @cached_property
- def _security_dependencies(self) -> list["Dependant"]:
- security_deps = [dep for dep in self.dependencies if dep._is_security_scheme]
- return security_deps
- @cached_property
- def is_gen_callable(self) -> bool:
- if self.call is None:
- return False # pragma: no cover
- if inspect.isgeneratorfunction(
- _impartial(self.call)
- ) or inspect.isgeneratorfunction(_unwrapped_call(self.call)):
- return True
- if inspect.isclass(_unwrapped_call(self.call)):
- return False
- dunder_call = getattr(_impartial(self.call), "__call__", None) # noqa: B004
- if dunder_call is None:
- return False # pragma: no cover
- if inspect.isgeneratorfunction(
- _impartial(dunder_call)
- ) or inspect.isgeneratorfunction(_unwrapped_call(dunder_call)):
- return True
- dunder_unwrapped_call = getattr(_unwrapped_call(self.call), "__call__", None) # noqa: B004
- if dunder_unwrapped_call is None:
- return False # pragma: no cover
- if inspect.isgeneratorfunction(
- _impartial(dunder_unwrapped_call)
- ) or inspect.isgeneratorfunction(_unwrapped_call(dunder_unwrapped_call)):
- return True
- return False
- @cached_property
- def is_async_gen_callable(self) -> bool:
- if self.call is None:
- return False # pragma: no cover
- if inspect.isasyncgenfunction(
- _impartial(self.call)
- ) or inspect.isasyncgenfunction(_unwrapped_call(self.call)):
- return True
- if inspect.isclass(_unwrapped_call(self.call)):
- return False
- dunder_call = getattr(_impartial(self.call), "__call__", None) # noqa: B004
- if dunder_call is None:
- return False # pragma: no cover
- if inspect.isasyncgenfunction(
- _impartial(dunder_call)
- ) or inspect.isasyncgenfunction(_unwrapped_call(dunder_call)):
- return True
- dunder_unwrapped_call = getattr(_unwrapped_call(self.call), "__call__", None) # noqa: B004
- if dunder_unwrapped_call is None:
- return False # pragma: no cover
- if inspect.isasyncgenfunction(
- _impartial(dunder_unwrapped_call)
- ) or inspect.isasyncgenfunction(_unwrapped_call(dunder_unwrapped_call)):
- return True
- return False
- @cached_property
- def is_coroutine_callable(self) -> bool:
- if self.call is None:
- return False # pragma: no cover
- if inspect.isroutine(_impartial(self.call)) and iscoroutinefunction(
- _impartial(self.call)
- ):
- return True
- if inspect.isroutine(_unwrapped_call(self.call)) and iscoroutinefunction(
- _unwrapped_call(self.call)
- ):
- return True
- if inspect.isclass(_unwrapped_call(self.call)):
- return False
- dunder_call = getattr(_impartial(self.call), "__call__", None) # noqa: B004
- if dunder_call is None:
- return False # pragma: no cover
- if iscoroutinefunction(_impartial(dunder_call)) or iscoroutinefunction(
- _unwrapped_call(dunder_call)
- ):
- return True
- dunder_unwrapped_call = getattr(_unwrapped_call(self.call), "__call__", None) # noqa: B004
- if dunder_unwrapped_call is None:
- return False # pragma: no cover
- if iscoroutinefunction(
- _impartial(dunder_unwrapped_call)
- ) or iscoroutinefunction(_unwrapped_call(dunder_unwrapped_call)):
- return True
- return False
- @cached_property
- def computed_scope(self) -> str | None:
- if self.scope:
- return self.scope
- if self.is_gen_callable or self.is_async_gen_callable:
- return "request"
- return None
|