Parallel APIs

Parallel processing APIs for mqdm.

mqdm.pool

pool(
    fn: Callable[..., R],
    iter: Iterator[Any]
    | list[Any]
    | tuple[Any, ...]
    | set[Any]
    | range,
    desc: str | DescFunc = "",
    bar_kw: dict[str, Any] | None = None,
    n_workers: int = 8,
    pool_mode: T_POOL_MODE = "process",
    results_: list[R] | None = None,
    ordered_: bool = True,
    squeeze_: bool = True,
    as_result_: bool = False,
    on_interrupt: Literal[
        "terminate", "kill"
    ] = "terminate",
    interrupt_grace: float = 2.0,
    runtime: Runtime | None = None,
    **kw: Any,
) -> list[R]

Collect ipool results into a list.

Parameters:

Name	Type	Description	Default
fn	Callable[..., R]	Function to call for each item.	required
iter	Iterator[Any] | list[Any] | tuple[Any, ...] | set[Any] | range	Items to process.	required
desc	str | DescFunc	Static description or callback used for the top-level progress bar.	''
bar_kw	dict[str, Any] | None	Extra keyword arguments forwarded to the top-level progress bar.	None
n_workers	int	Maximum number of workers to use.	8
pool_mode	T_POOL_MODE	Execution mode: "process", "thread", or "sequential".	'process'
results_	list[R] | None	Optional list to append results into.	None
ordered_	bool	Whether results should be returned in input order.	True
squeeze_	bool	Whether to reduce worker count for very small inputs.	True
runtime	Runtime | None	Runtime that should own the progress display.	None
**kw	Any	Extra keyword arguments forwarded to fn for every item.	{}

Returns:

Type	Description
list[R]	A list of collected results.

mqdm.apool async

apool(
    fn: Callable[..., R],
    iter: AsyncIterable[Any] | Iterable[Any],
    desc: str | DescFunc = "",
    bar_kw: dict[str, Any] | None = None,
    n_workers: int = 8,
    ordered_: bool = True,
    squeeze_: bool = True,
    as_result_: bool = False,
    on_error: OnError = "cancel",
    runtime: Runtime | None = None,
    **kw: Any,
) -> list[R]

Run async work over an iterable and return the results as a list.

The awaitable, list-collecting counterpart to :func:aipool — the same relation that :func:pool has to :func:ipool. Runs up to n_workers items concurrently as asyncio tasks (sync callables run via asyncio.to_thread).

Returns:

Type	Description
list[R]	A list of results (or :class:Result records if as_result_).

Example

results = await mqdm.apool(fetch, urls, n_workers=16, desc="fetching")

mqdm.aipool async

aipool(
    fn: Callable[..., R],
    iter: AsyncIterable[Any] | Iterable[Any],
    desc: str | DescFunc = "",
    bar_kw: dict[str, Any] | None = None,
    n_workers: int = 8,
    ordered_: bool = False,
    squeeze_: bool = True,
    as_result_: bool = False,
    on_error: OnError = "cancel",
    runtime: Runtime | None = None,
    **kw: Any,
) -> AsyncIterator[R]

Run async work over an iterable with bounded asyncio concurrency.

Parameters:

Name	Type	Description	Default
fn	Callable[..., R]	Async callable to run for each item. Sync callables are executed via asyncio.to_thread.	required
iter	AsyncIterable[Any] | Iterable[Any]	Items to process. Supports both sync and async iterables.	required
desc	str | DescFunc	Static description or callback used for the top-level progress bar.	''
bar_kw	dict[str, Any] | None	Extra keyword arguments forwarded to the top-level progress bar.	None
n_workers	int	Maximum number of in-flight asyncio tasks.	8
ordered_	bool	Whether results should be yielded in input order.	False
squeeze_	bool	Whether to reduce concurrency for very small known inputs.	True
as_result_	bool	Yield a :class:Result per task instead of raising task failures.	False
on_error	OnError	Error policy when as_result_ is false.	'cancel'
runtime	Runtime | None	Runtime that should own the progress display.	None
**kw	Any	Extra keyword arguments forwarded to fn for every item.	{}

Yields:

Type	Description
AsyncIterator[R]	Return values from fn (or :class:Result records if as_result_),
AsyncIterator[R]	as tasks complete.

Raises:

Type	Description
PoolError	If on_error='finish' and any task failed.

See also

:func:apool (awaitable list), and :func:ipool for thread/process pools.

mqdm.args

args(*a, **kw)

Bundle positional and keyword arguments for a single call.

Each item you pass to :func:mqdm.pool becomes one call to your function. Wrap an item in mqdm.args(...) when a call needs more than one argument, or a mix of positional and keyword arguments — otherwise the bare item is passed as the sole positional argument.

Example

# one worker call per args(...) bundle: work(path, out=..., overwrite=True)
jobs = [mqdm.args(p, out=f"{p}.done", overwrite=True) for p in paths]
mqdm.pool(work, jobs)

mqdm.Result dataclass

Result(
    index: int,
    arg: args,
    value: R | None = None,
    error: BaseException | None = None,
)

A single task's outcome: its input, return value, and any error.

index matches the task_index carried on the event stream, so a result correlates to its live events by key rather than by completion order.

mqdm.PoolError

PoolError(detail: str, results: list[Result])

Bases: Exception

Aggregates the failures from a pool run (on_error='finish').

A single distinct exception across many calls, or many distinct exceptions, all surface as one PoolError — there is no single "right" class to re-raise when tasks fail in different ways. The rendered message groups identical failures and lists the calls that produced them; results holds the failed :class:Result records (each with its index, arg and the original error) for programmatic inspection.

count property

count: int

Total number of failed calls.