hikari.internal.aio

Asyncio extensions and utilities.

Module Contents

async hikari.internal.aio.all_of(*aws, timeout=None)

Await the completion of all the given awaitable items.

If any fail or time out, then they are all cancelled.

Parameters:

*awstyping.Awaitable[T_co]

Awaitables to wait for.

timeouttyping.Optional[float]

Optional timeout to wait for, or None to not use one. If the timeout is reached, all awaitables are cancelled immediately.

Returns:

typing.Sequence[T_co]

The results of each awaitable in the order they were invoked in.

hikari.internal.aio.completed_future(result=None, /)

Create a future on the current running loop that is completed, then return it.

Parameters:

resultT

The value to set for the result of the future. T is a generic type placeholder for the type that the future will have set as the result. T may be None, in which case, this will return asyncio.Future[None].

Returns:

asyncio.Future[T]

The completed future.

Raises:

RuntimeError

When called in an environment with no running event loop.

hikari.internal.aio.destroy_loop(loop, logger)

Destroy the passed loop.

Parameters:

loopasyncio.AbstractEventLoop

The loop to destroy

loggerlogging.Logger

The logger to use for logging

async hikari.internal.aio.first_completed(*aws, timeout=None)

Wait for the first awaitable to complete.

The awaitables that don’t complete first will be cancelled.

Completion is defined as having a result or an exception set. Thus, cancelling any of the awaitables will also result in the others being cancelled.

If the first awaitable raises an exception, then that exception will be propagated.

Note

If more than one awaitable is completed before entering this call, then the first future is always returned.

Parameters:

*awstyping.Awaitable[typing.Any]

Awaitables to wait for.

timeouttyping.Optional[float]

Optional timeout to wait for, or None to not use one. If the timeout is reached, all awaitables are cancelled immediately.

hikari.internal.aio.get_or_make_loop()

Get the current usable event loop or create a new one.

Returns:

asyncio.AbstractEventLoop

The requested loop.

hikari.internal.aio.is_async_iterable(obj)

Determine if the object is an async iterable or not.

hikari.internal.aio.is_async_iterator(obj)

Determine if the object is an async iterator or not.