Using Python asyncio.gather for Concurrent Tasks

When you need to run several independent asynchronous operations at the same time in Python, asyncio.gather is the standard tool. It takes one or more awaitables, schedules them concurrently, and returns their results in the order they were passed. This article explains how it works, how to handle errors, and where its limits are.

What asyncio.gather Does

asyncio.gather accepts multiple awaitables—coroutines, tasks, or futures—and runs them as concurrent tasks on the event loop. The function returns a single awaitable that, when awaited, produces a list of results in the same order as the input awaitables.

import asyncio

async def fetch_data(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"Data from {name}"

async def main():
    results = await asyncio.gather(
        fetch_data("A", 0.2),
        fetch_data("B", 0.1),
        fetch_data("C", 0.3),
    )
    print(results)

asyncio.run(main())

The three fetch_data calls start together. Even though B finishes first, results is ['Data from A', 'Data from B', 'Data from C'] because gather preserves input order. This behavior is useful when the order of results must match the order of the calls, such as when combining data from several endpoints into a fixed sequence.

Passing Coroutines vs. Tasks

gather accepts coroutine objects directly. It wraps each coroutine in a Task automatically when scheduling it. You can also pass already-created Task objects, which is useful if you need to manage tasks separately.

async def main():
    task_a = asyncio.create_task(fetch_data("A", 0.2))
    task_b = asyncio.create_task(fetch_data("B", 0.1))
    results = await asyncio.gather(task_a, task_b)

Passing a coroutine object to gather is simpler for most cases. The main difference is that when you pass a coroutine, gather creates the Task internally; when you pass a Task, you already have a reference to it and can cancel or await it elsewhere. Avoid passing the same coroutine object to both create_task and gather: a coroutine object can be awaited only once.

Handling Exceptions with return_exceptions

By default, if any awaited coroutine raises an exception, gather propagates the first exception as soon as it occurs. The other coroutines are not cancelled: they continue running, but gather no longer collects their results. The exception is raised at the point where you await the gather call. If you passed Task objects, you can still await those tasks separately to get results from tasks that did not fail.

async def fail():
    raise RuntimeError("boom")

async def main():
    try:
        await asyncio.gather(fail(), fetch_data("A", 0.1))
    except RuntimeError as e:
        print(f"Caught: {e}")

If you want to collect successful results even when some coroutines fail, set return_exceptions=True. In that mode, gather returns a list where successful results are the returned values and failed coroutines are represented by the exception objects.

async def main():
    results = await asyncio.gather(
        fail(),
        fetch_data("A", 0.1),
        return_exceptions=True,
    )
    for r in results:
        if isinstance(r, Exception):
            print(f"Error: {r}")
        else:
            print(f"Success: {r}")

With return_exceptions=True, the exception is not raised; it is captured in the result list. This is useful when you want to handle partial failures without aborting the entire batch. The exception object itself is returned, so you can inspect it normally.

Cancellation Behavior

When the task that awaits gather is cancelled, cancellation propagates to the child coroutines gather is managing. Each child receives a CancelledError at its next await point, which can be important for group cleanup.

import asyncio

async def worker(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"Data from {name}"

async def run_group():
    await asyncio.gather(worker("A", 1), worker("B", 1))

async def main():
    task = asyncio.create_task(run_group())
    await asyncio.sleep(0.1)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Group cancelled")

asyncio.run(main())

If a child coroutine catches CancelledError and suppresses it, cancellation may not cleanly finish. In general, do not suppress CancelledError inside a coroutine. If you need cleanup, run it in a finally block and re-raise the cancellation.

When gather Is Not the Right Choice

gather is designed for running a fixed set of awaitables and collecting their results. If you need to process results as they complete, or if you have a dynamic stream of tasks, consider asyncio.as_completed or asyncio.wait.

as_completed yields futures in the order they finish, allowing you to handle each result immediately. wait gives you more control over waiting conditions (e.g., wait for the first exception or all tasks to complete).

For a simple batch where you need all results before proceeding, gather is the clearest option. If you have a list of coroutines created dynamically, you can still use gather by unpacking the list.

coros = [fetch_data(f"Item {i}", i * 0.1) for i in range(5)]
results = await asyncio.gather(*coros)

Avoiding Common Pitfalls

One frequent mistake is passing a list of coroutines without unpacking it. gather expects separate arguments, not a single list. Use the * operator to expand the list.

Another pitfall is mixing gather with blocking calls inside coroutines. If a coroutine calls a synchronous blocking function like time.sleep, it will block the entire event loop, defeating concurrency. Use asyncio.sleep or run blocking code in a thread pool with asyncio.to_thread (available in Python 3.9+).

# Bad: blocks the loop
async def bad_worker():
    time.sleep(1)

# Good: non-blocking sleep
async def good_worker():
    await asyncio.sleep(1)

Also be aware that gather does not limit concurrency. If you pass 1000 coroutines, all 1000 tasks are scheduled at once. For many I/O-bound operations this may be acceptable, but if you need to limit the number of simultaneous operations, use a semaphore or a task pool pattern.

Performance and Production Considerations

gather is efficient because it schedules tasks on the event loop without creating extra threads or processes. For I/O-bound workloads, this allows many concurrent operations with low overhead. However, the event loop runs on a single thread, so CPU-bound code will not benefit from gather alone. For CPU-bound work, use asyncio.run_in_executor or the multiprocessing module.

In production, monitor the number of tasks you create. If you have a very large batch, consider chunking the work to avoid memory pressure from holding many Task objects. Also, when using return_exceptions=True, the exception objects may keep traceback and frame references alive, so storing them long-term can increase memory usage.

Another production concern is timeouts. gather does not have a built-in timeout. If you need to bound the total execution time, wrap the gather call with asyncio.wait_for.

async def main():
    try:
        results = await asyncio.wait_for(
            asyncio.gather(fetch_data("A", 0.2), fetch_data("B", 0.3)),
            timeout=0.5,
        )
    except asyncio.TimeoutError:
        print("Timed out")

When a timeout occurs, wait_for cancels the gather task, which in turn cancels all child coroutines. This is a clean way to enforce an overall deadline.

Advanced Pattern: Gathering with a Semaphore

To limit concurrency while still using gather, you can wrap each coroutine with a semaphore. The semaphore ensures that only a limited number of tasks run at once, while gather still collects all results.

async def limited_fetch(sem: asyncio.Semaphore, name: str, delay: float) -> str:
    async with sem:
        return await fetch_data(name, delay)

async def main():
    sem = asyncio.Semaphore(3)
    coros = [limited_fetch(sem, f"Item {i}", i * 0.1) for i in range(10)]
    results = await asyncio.gather(*coros)

The semaphore is acquired before each fetch and released when the fetch completes. This pattern keeps the simplicity of gather while controlling resource usage. It is especially useful when hitting external APIs that have rate limits.

Compatibility Notes

asyncio.gather has been available since Python 3.4 and remains a common way to run a fixed set of awaitables. In Python 3.11+, asyncio.TaskGroup offers structured concurrency and may be preferable in new code, but gather still works on older Python versions.

One key difference is exception handling. TaskGroup cancels the remaining tasks when any task raises an exception other than CancelledError; gather with default settings propagates the first exception but lets the other tasks continue. If you want to collect successful results alongside failures, use gather(..., return_exceptions=True).