Use shared is_async_callable instead of inspect.iscoroutinefunction (#2389)

This commit is contained in:
Marcelo Trylesinski 2026-04-08 14:12:20 +02:00 • committed by GitHub
commit cf4e435db0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194

View file

@ -0,0 +1,33 @@
"""Callable inspection utilities.
Adapted from Starlette's `is_async_callable` implementation.
https://github.com/encode/starlette/blob/main/starlette/_utils.py
"""
from __future__ import annotations
import functools
import inspect
from collections.abc import Awaitable, Callable
from typing import Any, TypeGuard, TypeVar, overload
T = TypeVar("T")
AwaitableCallable = Callable[..., Awaitable[T]]
@overload
def is_async_callable(obj: AwaitableCallable[T]) -> TypeGuard[AwaitableCallable[T]]: ...
@overload
def is_async_callable(obj: Any) -> TypeGuard[AwaitableCallable[Any]]: ...
def is_async_callable(obj: Any) -> Any:
while isinstance(obj, functools.partial): # pragma: lax no cover
obj = obj.func
return inspect.iscoroutinefunction(obj) or (
callable(obj) and inspect.iscoroutinefunction(getattr(obj, "__call__", None))
)