hikari.internal.aio#
Asyncio extensions and utilities.
Module Contents#
- hikari.internal.aio.completed_future(result=None, /)[source]#
Create a future on the current running loop that is completed, then return it.
- Parameters
- result
T The value to set for the result of the future.
Tis a generic type placeholder for the type that the future will have set as the result.Tmay beNone, in which case, this will returnasyncio.Future[None].
- result
- Returns
asyncio.Future[T]The completed future.
- Raises
RuntimeErrorWhen called in an environment with no running event loop.
- hikari.internal.aio.is_async_iterator(obj)[source]#
Determine if the object is an async iterator or not.
- hikari.internal.aio.is_async_iterable(obj)[source]#
Determine if the object is an async iterable or not.
- async hikari.internal.aio.first_completed(*aws, timeout=None)[source]#
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
- *aws
typing.Awaitable[typing.Any] Awaitables to wait for.
- timeout
typing.Optional[float] Optional timeout to wait for, or
Noneto not use one. If the timeout is reached, all awaitables are cancelled immediately.
- *aws
- async hikari.internal.aio.all_of(*aws, timeout=None)[source]#
Await the completion of all the given awaitable items.
If any fail or time out, then they are all cancelled.
- Parameters
- *aws
typing.Awaitable[T_co] Awaitables to wait for.
- timeout
typing.Optional[float] Optional timeout to wait for, or
Noneto not use one. If the timeout is reached, all awaitables are cancelled immediately.
- *aws
- Returns
typing.Sequence[T_co]The results of each awaitable in the order they were invoked in.
- hikari.internal.aio.get_or_make_loop()[source]#
Get the current usable event loop or create a new one.
- Returns
asyncio.AbstractEventLoopThe requested loop.
- hikari.internal.aio.destroy_loop(loop, logger)[source]#
Destroy the passed loop.
- Parameters
- loop
asyncio.AbstractEventLoop The loop to destroy
- logger
logging.Logger The logger to use for logging
- loop