High-level Coroutine Utilities

A set of higher-level coroutine aggregation utilities based on Supervisor.

async as_completed_safe(coros: Iterable[Awaitable[aiotools.utils.T]], *, context: Optional[_contextvars.Context] = None) → AsyncGenerator[Awaitable[aiotools.utils.T], None]

This is a safer version of asyncio.as_completed() which uses aiotools.Supervisor as an underlying coroutine lifecycle keeper.

This requires Python 3.11 or higher to work properly with timeouts.

New in version 1.6.

Changed in version 2.0: It now uses aiotools.Supervisor internally and handles timeouts in a bettery way.

async gather_safe(coros: Iterable[Awaitable[aiotools.utils.T]], *, context: Optional[_contextvars.Context] = None) → List[Union[aiotools.utils.T, Exception]]

A safer version of asyncio.gather(). It wraps the passed coroutines with a Supervisor to ensure the termination of them when returned.

Additionally, it supports manually setting the context of each subtask.

Note that if it is cancelled from an outer scope (e.g., timeout), there is no way to retrieve partially completed or failed results. If you need to process them anyway, you must store the results in a separate place in the passed coroutines or use as_completed_safe() instead.

New in version 2.0.

async race(coros: Iterable[Awaitable[aiotools.utils.T]], *, continue_on_error: bool = False, context: Optional[_contextvars.Context] = None) → Tuple[aiotools.utils.T, Sequence[Exception]]

Returns the first result and cancelling all remaining coroutines safely. Passing an empty iterable of coroutines is not allowed.

If continue_on_error is set False (default), it will raise the first exception immediately, cancelling all remaining coroutines. This behavior is same to Javascript’s Promise.race(). The second item of the returned tuple is always empty.

If continue_on_error is set True, it will keep running until it encounters the first successful result. Then it returns the exceptions as a list in the second item of the returned tuple. If all coroutines fail, it will raise an ExceptionGroup to indicate the explicit failure of the entire operation.

You may use this function to implement a “happy eyeball” algorithm.

New in version 2.0.