sync

starlite.utils.sync.is_async_callable(value: Callable[[P], T]) → TypeGuard[Callable[[P], Awaitable[T]]]

Extend asyncio.iscoroutinefunction() to additionally detect async functools.partial() objects and class instances with async def __call__() defined.

Parameters:

value – Any

Returns:

Bool determining if type of value is an awaitable.

class starlite.utils.sync.AsyncCallable

Bases: Generic[P, T]

Wrap a callable into an asynchronous callable.

__init__(fn: Callable[[P], T]) → None

Initialize the wrapper from any callable.

Parameters:

fn – Callable to wrap - can be any sync or async callable.

async __call__(*args: ~typing.~P, **kwargs: ~typing.~P) → T

Proxy the wrapped function’s call method.

Parameters:

• *args – Args of the wrapped function.

• **kwargs – Kwargs of the wrapper function.

Returns:

The return value of the wrapped function.

starlite.utils.sync.as_async_callable_list(value: Union[Callable, List[Callable]]) → List[AsyncCallable]

Wrap callables in AsyncCallable s.

Parameters:

value – A callable or list of callables.

Returns:

A list of AsyncCallable instances

starlite.utils.sync.async_partial(fn: Callable) → Callable

Wrap a given sync function making it async.

In difference to the asyncio.run_sync() function, it allows for passing kwargs.

Parameters:

fn – A sync callable to wrap.

Returns:

A wrapper

class starlite.utils.sync.AsyncIteratorWrapper

Bases: Generic[T]

Asynchronous generator, wrapping an iterable or iterator.

__init__(iterator: Union[Iterator[T], Iterable[T]]) → None

Take a sync iterator or iterable and yields values from it asynchronously.

Parameters:

iterator – A sync iterator or iterable.