Python as_completed: Process Async Results as They Finish

python as_completed requires a clear understanding of the core syntax, runtime behavior, and practical implementation patterns demonstrated in the examples below.

When you run multiple coroutines with asyncio, you often need to act on each result as soon as it becomes available rather than waiting for the slowest task. The asyncio.as_completed function provides this behavior by returning an iterator that yields awaitables in the order their underlying tasks complete. This is particularly useful when you have a batch of independent I/O-bound operations and want to process results incrementally.

How as_completed Works

asyncio.as_completed takes an iterable of awaitables (coroutines, tasks, or futures) and returns an iterator of awaitables. Each time you await the next item from the iterator, you get the result of the task that finished next. The iterator does not yield results directly; you must await each yielded object to retrieve the actual result or raise an exception.

Here is a minimal example:

import asyncio

async def worker(name, delay):
    await asyncio.sleep(delay)
    return f"{name} done after {delay}s"

async def main():
    tasks = [
        asyncio.create_task(worker("A", 3)),
        asyncio.create_task(worker("B", 1)),
        asyncio.create_task(worker("C", 2)),
    ]
    for coro in asyncio.as_completed(tasks):
        result = await coro
        print(result)

asyncio.run(main())

The output will be:

B done after 1s
C done after 2s
A done after 3s

The order depends on completion time, not the order in the input list.

Basic Usage: Awaiting Results as They Arrive

A common pattern is to start several tasks and process each result as soon as it is ready. This avoids blocking on the slowest task before handling faster ones. For example, if you are fetching data from multiple endpoints, you can update a UI or write to a stream as each response arrives.

async def fetch(url):
    # Simulate network latency
    await asyncio.sleep(1)
    return f"Data from {url}"

async def main():
    urls = ["https://example.com/1", "https://example.com/2", "https://example.com/3"]
    tasks = [asyncio.create_task(fetch(url)) for url in urls]
    for coro in asyncio.as_completed(tasks):
        data = await coro
        print(data)

In this example, all three requests start concurrently, and you print each result as soon as its simulated delay completes.

Handling Exceptions with as_completed

When a task raises an exception, that exception is raised when you await the corresponding yielded object. You can catch it per task, which is useful when you want to continue processing other tasks even if one fails.

async def risky(delay, should_fail):
    await asyncio.sleep(delay)
    if should_fail:
        raise RuntimeError("Task failed")
    return f"Success after {delay}s"

async def main():
    tasks = [
        asyncio.create_task(risky(1, False)),
        asyncio.create_task(risky(2, True)),
        asyncio.create_task(risky(3, False)),
    ]
    for coro in asyncio.as_completed(tasks):
        try:
            result = await coro
        except RuntimeError as e:
            print(f"Caught error: {e}")
        else:
            print(f"Result: {result}")

The loop continues even if one task fails. If you do not catch the exception, it will propagate and stop the loop, potentially leaving other tasks unawaited.

as_completed vs asyncio.gather

asyncio.gather waits for all tasks to complete and returns a list of results in the original order. as_completed yields results as they finish. The table below summarizes the key differences.

Choose gather when you need the full set of results before proceeding. Use as_completed when you want to start processing early or when the order of completion matters.

Controlling Concurrency and Timeouts

as_completed does not limit how many tasks run at once. If you need to cap concurrency, combine it with a semaphore or create tasks in batches. For timeouts, wrap each awaited coroutine with asyncio.wait_for.

async def main():
    sem = asyncio.Semaphore(2)

    async def limited_task(i):
        async with sem:
            await asyncio.sleep(1)
            return i

    tasks = [asyncio.create_task(limited_task(i)) for i in range(5)]
    for coro in asyncio.as_completed(tasks):
        try:
            result = await asyncio.wait_for(coro, timeout=2)
        except asyncio.TimeoutError:
            print("Task timed out")
        else:
            print(f"Result: {result}")

In Python 3.11+, you can also use asyncio.timeout as a context manager for a group of tasks.

Common Pitfalls and Edge Cases

• Empty input: as_completed([]) returns an empty iterator, so the loop body never runs.
• Consume the iterator: If you create tasks but never iterate over the as_completed object, the tasks still run but their results are discarded. Always consume the iterator to avoid unhandled exceptions.
• Task cancellation: If a task is cancelled, awaiting its yielded object raises asyncio.CancelledError. Handle it explicitly if cancellation is part of your design.
• Input order vs completion order: The iterator does not preserve the order of the input list. Rely on completion order only.

Practical Example: Processing HTTP Requests

A realistic use case is fetching multiple URLs with aiohttp. The pattern is the same as the simulated example above.

import aiohttp
import asyncio

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = ["https://example.com", "https://example.org", "https://example.net"]
    async with aiohttp.ClientSession() as session:
        tasks = [asyncio.create_task(fetch(session, url)) for url in urls]
        for coro in asyncio.as_completed(tasks):
            try:
                text = await coro
            except Exception as e:
                print(f"Failed to fetch: {e}")
            else:
                print(f"Got {len(text)} bytes")

asyncio.run(main())

This pattern lets you start all requests concurrently and handle each response as soon as it arrives, which is ideal for streaming or progressive display.