Skip to content

Processing Results in Completion Order with as_completed

asyncio.gather() hands you every result at once, after the slowest task has finished. When the results are independent — rows to write, pages to parse, notifications to send — that wastes the time between the first result and the last. asyncio.as_completed() yields them in the order they finish. In a test with 20 tasks taking 10–200 ms, gather made the first result available at 202 ms; as_completed delivered it at 10 ms. The API had two long-standing warts — you could not tell which input a result came from, and its timeout leaves work running — and Python 3.13 fixed the first. This guide uses the modern form and works around the second.

Prerequisites

1. Use the async-for form on 3.13+

Since Python 3.13, as_completed() returns an object that supports async for, and each iteration yields the original task — the same object you passed in — once it is done:

import asyncio


async def fetch(n: float) -> float:
    await asyncio.sleep(n)
    return n


async def main() -> None:
    tasks = [asyncio.create_task(fetch(d), name=f"job-{d}") for d in (0.3, 0.1, 0.2)]
    async for task in asyncio.as_completed(tasks):
        print(task.get_name(), task.result())   # job-0.1, job-0.2, job-0.3


asyncio.run(main())

Because the yielded object is your task, you can attach the input to it — a name, a dict keyed by task, an attribute — and recover it when the result arrives. Verified on 3.14: every yielded object satisfied task in tasks.

In the older plain-for form, each iteration yields a new coroutine that resolves to the next result, and that coroutine is not one of your tasks (f in tasks was False in the same test). Mapping a result back to its input needed the input embedded in the result itself.

Verify: print task in tasks inside the loop; it must be True for the async-for form.

When the first result becomes usable 3 lanes over time. When the first result becomes usable tasks 20 tasks finishing between 10 ms and 200 ms gather waiting for the slowest all 20 as_completed first one result every ~10 ms time → Same 20 tasks; gather waits for the slowest, as_completed hands each result over as soon as it exists.

2. Map results back to inputs

With the task in hand, keep a dictionary from task to input. This is the pattern for anything where the result alone does not identify its source:

async def check_all(urls: list[str], client) -> dict[str, int]:
    tasks = {asyncio.create_task(client.head(u)): u for u in urls}
    status: dict[str, int] = {}
    async for task in asyncio.as_completed(tasks):
        url = tasks[task]
        if task.exception() is not None:
            log.warning("%s failed: %r", url, task.exception())
            continue
        status[url] = task.result().status_code
        progress.update(len(status), total=len(urls))   # progress as results land
    return status

On Python 3.11 and 3.12, wrap each coroutine so it returns its input alongside its result:

async def tagged(key, coro):
    try:
        return key, await coro, None
    except Exception as exc:
        return key, None, exc

for fut in asyncio.as_completed([tagged(u, client.head(u)) for u in urls]):
    url, resp, exc = await fut

The wrapper catching exceptions matters in the old form: an exception raised from await fut tells you nothing about which input failed.

Verify: introduce one failing URL; the log line must name it.

3. Handle exceptions per result, not per batch

as_completed never raises a task's exception on its own. In the async-for form you inspect task.exception(); in the old form the exception is raised by await fut, for that one result only, and the loop can continue:

async for task in asyncio.as_completed(tasks):
    try:
        row = task.result()          # raises this task's exception, if any
    except (TimeoutError, ConnectionError) as exc:
        failures.append((tasks[task], exc))
        continue
    await sink.write(row)

This is the main behavioural difference from gather and TaskGroup: one failure does not abort the others. That is what you want for independent work. For work where one failure should stop everything, use a TaskGroup and accept that results are only available at the end.

Verify: with one failing task among many, the loop processes every successful result and records exactly one failure.

as_completed against gather and TaskGroup A grid of 4 rows by 4 columns. as_completed against gather and TaskGroup property as_completed gather TaskGroup first result usable when it finishes when all finish when all finish one task fails only that result raises first error raised all cancelled timeout TimeoutError, work keeps running via wait_for via asyncio.timeout maps to input 3.13+: yields your task by position you hold tasks as_completed is the streaming option, with per-result errors and no automatic cleanup.

4. Do not trust the timeout to stop work

as_completed(tasks, timeout=...) raises TimeoutError from the next iteration once the deadline passes. It does not cancel the tasks that have not finished. Verified: with tasks of 50 ms and 500 ms and a 100 ms timeout, the loop raised after the first result, and the 500 ms task was still running afterwards.

Use asyncio.timeout() around the loop and cancel explicitly in finally:

async def collect_until(tasks: list[asyncio.Task], deadline_s: float) -> list:
    results = []
    try:
        async with asyncio.timeout(deadline_s):
            async for task in asyncio.as_completed(tasks):
                if task.exception() is None:
                    results.append(task.result())
    except TimeoutError:
        log.info("deadline hit with %d/%d results", len(results), len(tasks))
    finally:
        for t in tasks:
            t.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)
    return results

This is a partial-results fan-out: return whatever arrived before the deadline, abandon the rest cleanly. The same shape with exception groups is in returning partial results from a fan-out before a deadline.

Verify: after collect_until returns, every task in the list is done().

5. Bound the fan-out

as_completed starts nothing itself; every task you pass in is already running. With 10,000 inputs that means 10,000 concurrent requests. Bound the concurrency inside each task with a semaphore, so tasks queue for a slot instead of all hitting the backend:

async def bounded_map(fn, items, limit: int = 32):
    sem = asyncio.Semaphore(limit)

    async def run(item):
        async with sem:
            return await fn(item)

    tasks = {asyncio.create_task(run(i)): i for i in items}
    try:
        async for task in asyncio.as_completed(tasks):
            yield tasks[task], task
    finally:
        for t in tasks:
            t.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)

For very large inputs, creating one task per item still costs memory — about 1.3 KB per pending task including its coroutine frame, measured with tracemalloc on 3.14. Beyond tens of thousands of items, switch to a fixed set of workers pulling from a queue, as in building an async worker pool with TaskGroup.

Verify: the backend never sees more than limit concurrent requests, and breaking out of the consumer's loop cancels everything still queued.

Bounded fan-out with streaming results A flow of 4 stages. Bounded fan-out with streaming results one task per item all created up front Semaphore(limit) at most limit run as_completed yield in finish order finally cancel the rest The semaphore bounds load on the backend; the finally bounds the work's lifetime to the consumer's.

Verification

Completion-order processing is correct when:

  • The first result is handled at the speed of the fastest task, not the slowest.
  • Each result can be traced to its input, including failures.
  • One failure does not stop the others, unless you want it to.
  • No task outlives the loop — after a timeout, an exception, or an early break.

Diagnostic Hook: record time-to-first-result and time-to-last-result per batch. A large gap between them is the reason to use as_completed at all; if the gap is small, gather is simpler. A time-to-last-result that grows while time-to-first stays flat points at a slow tail in the backend — a candidate for hedging.

Pitfalls & edge cases

  • Relying on the plain-for form returning your tasks. It does not; only the 3.13+ async-for form yields the originals.
  • Passing coroutines and expecting order. as_completed wraps them in tasks you never see; you lose the mapping and the ability to cancel.
  • break without cleanup. Breaking out of the loop leaves the remaining tasks running.
  • Using the timeout parameter as a deadline on the work. It is a deadline on the waiting.
  • Unbounded input. Every item becomes a running task immediately; bound it.

Frequently Asked Questions

How do I process asyncio results as they complete?

Pass your tasks to asyncio.as_completed and iterate. On Python 3.13 and later use async for, which yields each original task when it is done, so you can call task.result() and map it back to its input.

Does asyncio.as_completed cancel remaining tasks on timeout?

No. It raises TimeoutError from the next iteration but the unfinished tasks keep running. Wrap the loop in asyncio.timeout and cancel the tasks in a finally block.

How do I know which input an as_completed result belongs to?

On 3.13+, the async-for form yields your own task objects, so keep a dict from task to input. On earlier versions, wrap each coroutine to return its input together with its result.

Is as_completed faster than gather?

The total time is the same, set by the slowest task. The difference is when the first result is usable: in a 20-task test, as_completed delivered the first after 10 ms and gather after 202 ms.