hikari.internal.aio#
Asyncio extensions and utilities.
Module Contents#
- 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
None
to 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.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.
T
is a generic type placeholder for the type that the future will have set as the result.T
may beNone
, in which case, this will returnasyncio.Future[None]
.
- result
- 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)[source]#
Destroy the passed loop.
- Parameters:
- loop
asyncio.AbstractEventLoop
The loop to destroy
- logger
logging.Logger
The logger to use for logging
- loop
- 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
None
to not use one. If the timeout is reached, all awaitables are cancelled immediately.
- *aws
- hikari.internal.aio.get_or_make_loop()[source]#
Get the current usable event loop or create a new one.
- Returns:
asyncio.AbstractEventLoop
The requested loop.