Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
49 changes: 35 additions & 14 deletions fastapi/dependencies/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -77,36 +77,57 @@ def _uses_scopes(self) -> bool:

@cached_property
def _unwrapped_call(self) -> Any:
call = self.call
if call is None:
return call # pragma: no cover
while isinstance(call, partial):
call = call.func
return call

@cached_property
def _fully_unwrapped_call(self) -> Any:
if self.call is None:
return self.call # pragma: no cover
unwrapped = inspect.unwrap(self.call)
if isinstance(unwrapped, partial):
unwrapped = unwrapped.func
return unwrapped

@cached_property
def is_gen_callable(self) -> bool:
if inspect.isgeneratorfunction(self._unwrapped_call):
@staticmethod
def _check_gen(call: Any) -> bool:
if inspect.isgeneratorfunction(call):
return True
dunder_call = getattr(self._unwrapped_call, "__call__", None) # noqa: B004
dunder_call = getattr(call, "__call__", None) # noqa: B004
return inspect.isgeneratorfunction(dunder_call)

@cached_property
def is_async_gen_callable(self) -> bool:
if inspect.isasyncgenfunction(self._unwrapped_call):
@staticmethod
def _check_async_gen(call: Any) -> bool:
if inspect.isasyncgenfunction(call):
return True
dunder_call = getattr(self._unwrapped_call, "__call__", None) # noqa: B004
dunder_call = getattr(call, "__call__", None) # noqa: B004
return inspect.isasyncgenfunction(dunder_call)

@cached_property
def is_coroutine_callable(self) -> bool:
if inspect.isroutine(self._unwrapped_call):
return iscoroutinefunction(self._unwrapped_call)
if inspect.isclass(self._unwrapped_call):
@staticmethod
def _check_coroutine(call: Any) -> bool:
if inspect.isroutine(call):
return iscoroutinefunction(call)
if inspect.isclass(call):
return False
dunder_call = getattr(self._unwrapped_call, "__call__", None) # noqa: B004
dunder_call = getattr(call, "__call__", None) # noqa: B004
return iscoroutinefunction(dunder_call)

@cached_property
def is_gen_callable(self) -> bool:
return self._check_gen(self._unwrapped_call) or self._check_gen(self._fully_unwrapped_call)

@cached_property
def is_async_gen_callable(self) -> bool:
return self._check_async_gen(self._unwrapped_call) or self._check_async_gen(self._fully_unwrapped_call)

@cached_property
def is_coroutine_callable(self) -> bool:
return self._check_coroutine(self._unwrapped_call) or self._check_coroutine(self._fully_unwrapped_call)

@cached_property
def computed_scope(self) -> Union[str, None]:
if self.scope:
Expand Down