duck.contrib.asyncio

Utilities for managing and executing coroutines on available asyncio event loops.

This module provides functions for submitting coroutines to an available asyncio event loop, which is useful in applications where multiple execution contexts (e.g., ASGI, WSGI) might be running. The utility functions ensure compatibility with both synchronous and asynchronous workflows.

Functions:

• run_on_available_loop: Submits a coroutine to the currently available event loop.

• get_available_event_loop: Retrieves an appropriate event loop depending on the application’s context.

Module Contents

Functions

get_available_event_loop	Retrieve an available asyncio event loop depending on the application’s execution context.
run_on_available_loop	Submit an asynchronous coroutine to an available asyncio event loop.

API

duck.contrib.asyncio.get_available_event_loop(id: Optional[str] = None) → asyncio.AbstractEventLoop

Retrieve an available asyncio event loop depending on the application’s execution context.

This function determines the appropriate asyncio event loop to use based on the application’s configuration. For example: - In an ASGI or WSGI context, it retrieves a background event from AsyncioLoopManager.

Returns:

The available asyncio event loop.

Return type:

asyncio.AbstractEventLoop

Raises:

• RuntimeError – If no loop is found.

• ManagerNotFound – If asyncio loop manager with the provided id is not found.

… rubric:: Example

loop = get_available_event_loop() print(loop.is_running()) # Check if the event loop is running

duck.contrib.asyncio.run_on_available_loop(coro: Coroutine, id: Optional[str] = None, return_sync_future: bool = False) → Union[duck.utils.asyncio.eventloop.SyncFuture, asyncio.Future]

Submit an asynchronous coroutine to an available asyncio event loop.

This function ensures that the provided coroutine is executed on an appropriate asyncio event loop. It supports both synchronous and asynchronous workflows by optionally wrapping the result in a SyncFuture.

Parameters:

• coro – The coroutine to schedule for execution.

• id – The asyncio loop manager ID.

• return_sync_future – If True, wraps the result in a SyncFuture to allow blocking operations in synchronous contexts.

Returns:

• If return_sync_future is False, returns an asyncio.Future representing the scheduled coroutine.

• If return_sync_future is True, returns a SyncFuture that can be used for blocking operations.

Return type:

Union[SyncFuture, asyncio.Future]

Raises:

RuntimeError – If no event loop is available or the event loop is not currently running.

… rubric:: Example

async def my_coroutine(): return “Hello, asyncio!” future = run_on_available_loop(my_coroutine()) print(future.result()) # Blocks until the coroutine completes