AsyncExecutor API¶

The antflow.executor module provides a familiar interface for concurrent async execution, modeled after Python's concurrent.futures.

Overview¶

The AsyncExecutor manages a pool of workers to execute async functions concurrently. It is ideal for simple parallel processing tasks where you don't need the full complexity of a multi-stage pipeline.

Key features:

submit(): Schedule a single task.
map(): Apply a function to an iterable and return results as a list.
map_iter(): Apply a function to an iterable, yielding results (async iterator).
as_completed(): Iterate over futures as they finish.
wait(): Wait for a collection of futures with flexible conditions.

Usage Example¶

import asyncio
from antflow import AsyncExecutor

async def process_item(x):
    await asyncio.sleep(0.1)
    return x * 2

async def main():
    # Use as a context manager
    async with AsyncExecutor(max_workers=5) as executor:

        # 1. Submit a single task
        future = executor.submit(process_item, 10)
        result = await future.result()
        print(f"Result: {result}")

        # 2. Map over a list - returns list directly
        results = await executor.map(process_item, range(5))
        print(f"Results: {results}")

        # 3. Map with streaming (async iterator)
        async for res in executor.map_iter(process_item, range(5)):
            print(f"Streamed: {res}")

asyncio.run(main())

Class Reference¶

AsyncFuture¶

AsyncFuture ¶

AsyncFuture(sequence_id: int)

Bases: Generic[R]

An async-compatible future that holds the result of an async task. Similar to concurrent.futures.Future but for asyncio.

set_result ¶

set_result(result: R) -> None

Set the result and mark the future as done.

set_exception ¶

set_exception(exception: Exception) -> None

Set an exception and mark the future as done.

result `async` ¶

result(timeout: float | None = None) -> R

Wait for and return the result.

Parameters:

Name	Type	Description	Default
`timeout`	`float \| None`	Maximum time to wait in seconds	`None`

Returns:

Type	Description
`R`	The task result

Raises:

Type	Description
`TimeoutError`	If timeout is exceeded
`Exception`	The exception set by set_exception()

done ¶

done() -> bool

Return True if the future is done.

exception ¶

exception() -> Exception | None

Return the exception set on this future, or None.

AsyncExecutor¶

AsyncExecutor ¶

AsyncExecutor(max_workers: int = 5)

An async executor with concurrent.futures-style API. Manages a pool of workers that execute async tasks concurrently.

Initialize the executor.

Parameters:

Name	Type	Description	Default
`max_workers`	`int`	Maximum number of concurrent workers	`5`

submit ¶

submit(fn: Callable[..., Awaitable[R]], *args: Any, retries: int = 0, retry_delay: float = 0.1, semaphore: Semaphore | None = None, **kwargs: Any) -> AsyncFuture[R]

Submit a task for execution.

Parameters:

Name	Type	Description	Default
`fn`	`Callable[..., Awaitable[R]]`	Async callable to execute	required
`*args`	`Any`	Positional arguments for fn	`()`
`retries`	`int`	Number of retries on failure	`0`
`retry_delay`	`float`	Delay between retries in seconds	`0.1`
`semaphore`	`Semaphore \| None`	Optional semaphore to limit concurrency	`None`
`**kwargs`	`Any`	Keyword arguments for fn	`{}`

Returns:

Type	Description
`AsyncFuture[R]`	AsyncFuture that can be awaited for the result

Raises:

Type	Description
`ExecutorShutdownError`	If executor has been shut down

map `async` ¶

map(fn: Callable[..., Awaitable[R]], *iterables: Iterable[T], timeout: float | None = None, retries: int = 0, retry_delay: float = 0.1) -> List[R]

Map an async function over iterables and return results as a list.

Similar to concurrent.futures.Executor.map(), but returns a list directly instead of an iterator (since list(executor.map(...)) is the common pattern).

Parameters:

Name	Type	Description	Default
`fn`	`Callable[..., Awaitable[R]]`	Async callable to map	required
`*iterables`	`Iterable[T]`	Iterables to map over	`()`
`timeout`	`float \| None`	Maximum time to wait for each result	`None`
`retries`	`int`	Number of retries on failure	`0`
`retry_delay`	`float`	Delay between retries in seconds	`0.1`

Returns:

Type	Description
`List[R]`	List of results from fn applied to each input, in input order

Raises:

Type	Description
`ExecutorShutdownError`	If executor has been shut down

Example

async with AsyncExecutor(max_workers=5) as executor:
    results = await executor.map(process, range(100))
    print(results)  # [0, 2, 4, 6, ...]

map_iter `async` ¶

map_iter(fn: Callable[..., Awaitable[R]], *iterables: Iterable[T], timeout: float | None = None, retries: int = 0, retry_delay: float = 0.1) -> AsyncIterator[R]

Map an async function over iterables, yielding results in input order.

Use this instead of map() when you need streaming behavior: - Process results as they arrive - Handle large datasets without loading all results into memory - Early exit when a condition is met

Parameters:

Name	Type	Description	Default
`fn`	`Callable[..., Awaitable[R]]`	Async callable to map	required
`*iterables`	`Iterable[T]`	Iterables to map over	`()`
`timeout`	`float \| None`	Maximum time to wait for each result	`None`
`retries`	`int`	Number of retries on failure	`0`
`retry_delay`	`float`	Delay between retries in seconds	`0.1`

Yields:

Type	Description
`AsyncIterator[R]`	Results from fn applied to each input

Raises:

Type	Description
`ExecutorShutdownError`	If executor has been shut down

Example

async with AsyncExecutor(max_workers=5) as executor:
    async for result in executor.map_iter(process, range(100)):
        print(result)
        if result > 50:
            break  # Early exit

as_completed `async` ¶

as_completed(futures: list[AsyncFuture[R]], timeout: float | None = None) -> AsyncIterator[AsyncFuture[R]]

Yield futures as they complete.

Parameters:

Name	Type	Description	Default
`futures`	`list[AsyncFuture[R]]`	List of futures to wait for	required
`timeout`	`float \| None`	Maximum time to wait for all futures	`None`

Yields:

Type	Description
`AsyncIterator[AsyncFuture[R]]`	Futures as they complete

Raises:

Type	Description
`TimeoutError`	If timeout is exceeded

wait `async` ¶

wait(futures: Iterable[AsyncFuture[R]], timeout: float | None = None, return_when: WaitStrategy = WaitStrategy.ALL_COMPLETED) -> Tuple[Set[AsyncFuture[R]], Set[AsyncFuture[R]]]

Wait for futures to complete with different strategies.

Similar to concurrent.futures.wait() but for async operations.

Parameters:

Name	Type	Description	Default
`futures`	`Iterable[AsyncFuture[R]]`	Iterable of AsyncFuture objects to wait for	required
`timeout`	`float \| None`	Maximum time to wait in seconds	`None`
`return_when`	`WaitStrategy`	Strategy for when to return: - FIRST_COMPLETED: Return when any future completes - FIRST_EXCEPTION: Return when any future raises an exception - ALL_COMPLETED: Return when all futures complete (default)	`ALL_COMPLETED`

Returns:

Type	Description
`Tuple[Set[AsyncFuture[R]], Set[AsyncFuture[R]]]`	Tuple of (done, not_done) sets of futures

Raises:

Type	Description
`TimeoutError`	If timeout is exceeded (with ALL_COMPLETED)

Example

futures = [executor.submit(task, i) for i in range(10)]
done, pending = await executor.wait(
    futures,
    return_when=WaitStrategy.FIRST_EXCEPTION
)

shutdown `async` ¶

shutdown(wait: bool = True, cancel_futures: bool = False) -> None

Shut down the executor.

Parameters:

Name	Type	Description	Default
`wait`	`bool`	If True, wait for all pending tasks to complete. If False, the stop signal is set without draining the queue, so tasks still queued (and not picked up before workers exit) may be left unprocessed. Use cancel_futures=True to fail them explicitly.	`True`
`cancel_futures`	`bool`	If True, cancel all pending futures	`False`

aenter `async` ¶

__aenter__() -> 'AsyncExecutor'

Context manager entry.

aexit `async` ¶

__aexit__(_exc_type: type[BaseException] | None, _exc_val: BaseException | None, _exc_tb: TracebackType | None) -> None

Context manager exit with automatic shutdown.

AsyncExecutor API¶

Overview¶

Usage Example¶

Class Reference¶

AsyncFuture¶

AsyncFuture ¶

set_result ¶

set_exception ¶

result async ¶

done ¶

exception ¶

AsyncExecutor¶

AsyncExecutor ¶

submit ¶

map async ¶

map_iter async ¶

as_completed async ¶

wait async ¶

shutdown async ¶

__aenter__ async ¶

__aexit__ async ¶

result `async` ¶

map `async` ¶

map_iter `async` ¶

as_completed `async` ¶

wait `async` ¶

shutdown `async` ¶

aenter `async` ¶

aexit `async` ¶