Python as_completed: Process Async Results as They Finish

asyncio.as_completed returns an iterator of awaitables in the order they finish. It is useful when you run multiple coroutines and want to act on each result as soon as it becomes available, rather than waiting for the slowest task.

How as_completed Works

asyncio.as_completed takes an iterable of awaitables (coroutines, tasks, or futures) and returns an iterator of awaitables. The iterator only yields an awaitable after its underlying task has finished; you must still 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 follows completion order, not the order of the input list:

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

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 per-task timeouts, wrap the original coroutine with asyncio.wait_for when creating the task. Wrapping the object yielded by as_completed with wait_for is not a useful timeout because the yielded awaitable is already finished.

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(asyncio.wait_for(limited_task(i), timeout=2))
        for i in range(5)
    ]
    for coro in asyncio.as_completed(tasks):
        try:
            result = await coro
        except asyncio.TimeoutError:
            print("Task timed out")
        else:
            print(f"Result: {result}")

If you want one deadline for the whole iteration, pass timeout to as_completed:

try:
    for coro in asyncio.as_completed(tasks, timeout=2):
        result = await coro
        print(result)
except asyncio.TimeoutError:
    print("Timed out waiting for remaining tasks")

In Python 3.11+, you can also use asyncio.timeout as a context manager around the whole loop.

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 you will not observe their results or exceptions. Always consume the iterator or retrieve results from the task objects explicitly.
• 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.