Skip to content

call_or_await

fast_depends.utils.run_async async #

run_async(
    func: Union[
        Callable[P, T], Callable[P, Awaitable[T]]
    ],
    *args: args,
    **kwargs: kwargs
) -> T
Source code in fast_depends/utils.py 
39
40
41
42
43
44
45
46
47
48
49
50
async def run_async(
    func: Union[
        Callable[P, T],
        Callable[P, Awaitable[T]],
    ],
    *args: P.args,
    **kwargs: P.kwargs,
) -> T:
    if is_coroutine_callable(func):
        return await cast(Callable[P, Awaitable[T]], func)(*args, **kwargs)
    else:
        return await run_in_threadpool(cast(Callable[P, T], func), *args, **kwargs)