Function run_in_threadpool

starlette/concurrency.py:32–34

(func: Callable[P, T], *args: P.args, **kwargs: P.kwargs)

async def run_in_threadpool(func: Callable[P, T], *args: P.args, **kwargs: P.kwargs) -> T:
    func = functools.partial(func, *args, **kwargs)
    return await anyio.to_thread.run_sync(func)


class _StopIteration(Exception):