asynchronous

Utilities for async programming.

import adulib.asynchronous as this_module

is_in_event_loop

is_in_event_loop()

---

batch_executor (async)

batch_executor(
   func: Callable,
   constant_kwargs: Dict[str, Any],
   batch_args: Optional[Iterable[Tuple[Any, ...]]],
   batch_kwargs: Optional[Iterable[Dict[str, Any]]],
   concurrency_limit: Optional[int],
   verbose: bool,
   progress_bar_desc: str
)

Executes a batch of asynchronous tasks.

Parameters: - func (Callable): The asynchronous function to execute for each batch. - constant_kwargs (Dict[str, Any], optional): Constant keyword arguments to pass to each function call. - batch_args (Optional[Iterable[Tuple[Any, …]]], optional): Iterable of argument tuples for each function call. - batch_kwargs (Optional[Iterable[Dict[str, Any]]], optional): Iterable of keyword argument dictionaries for each function call. - concurrency_limit (Optional[int], optional): Maximum number of concurrent tasks. If None, no limit is applied. - verbose (bool, optional): If True, displays a progress bar. Default is True. - progress_bar_desc (str, optional): Description for the progress bar. Default is “Processing”.

Returns: - List of results from the executed tasks.

Raises: - ValueError: If both ‘batch_args’ and ‘batch_kwargs’ are empty or if their lengths do not match.

---

async def sample_function(x, y, z):
    await asyncio.sleep(0.1)
    return z*(x + y)

constant_kwargs = {'z': 10}
batch_args = [(1,), (3,), (5,)]
batch_kwargs = [{'y': 2}, {'y': 4}, {'y': 6}]

results = await batch_executor(
    func=sample_function,
    constant_kwargs=constant_kwargs,
    batch_args=batch_args,
    batch_kwargs=batch_kwargs,
    concurrency_limit=2,
    verbose=False,
)

print("Results:", results)

Results: [30, 70, 110]