duck.utils.asyncio

Asyncio utilities and helpers.

Submodules

Package Contents

Functions

create_task

Create an asyncio task and handle exceptions, optionally ignoring specified errors.

in_async_context

Check if the current code is running inside an asynchronous context.

API

duck.utils.asyncio.create_task(coro: Coroutine, on_complete: Optional[Callable[[asyncio.Task], None]] = None, raise_on_exception: bool = True, ignore_errors: Optional[List[Type[BaseException]]] = None, loop: Optional = None) → asyncio.Task[source]

Create an asyncio task and handle exceptions, optionally ignoring specified errors.

Parameters:

  • coro – The coroutine to run.

  • on_complete – Called with the task when finished.

  • raise_on_exception – Raises exceptions from the task unless ignored.

  • ignore_errors – Exception types to ignore.

  • loop – Custom event loop to use for creating task.

Raises:

  • RuntimeError – If the loop is provided and is not running.

  • Exception – If not ignored and raise_on_exception is True.

duck.utils.asyncio.in_async_context() → bool[source]

Check if the current code is running inside an asynchronous context.

Returns:

True if called within an async def coroutine (i.e., there’s a running event loop), False otherwise.

Return type:

bool

Example:

>>> in_async_context()
False

>>> async def main():
...     print(in_async_context())
>>> asyncio.run(main())
True