Using Python asyncio.gather for Concurrent Tasks

When you need to run several independent async 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 python asyncio gather 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. Be careful not to pass a coroutine that is already scheduled as a task—that would run it twice.

Handling Exceptions with return_exceptions

By default, if any awaited coroutine raises an exception, gather propagates that exception immediately, and the other coroutines continue running but their results are lost. The exception is raised at the point where you await the gather call.

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. Note that the exception object itself is returned, not the traceback, but you can inspect it normally.

Cancellation Behavior

When the task that awaits gather is cancelled, the cancellation propagates to all the child coroutines that gather is managing. This is important for cleanup: if you cancel a group of operations, each coroutine gets a CancelledError at its next await point.

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

If a coroutine catches CancelledError and suppresses it, gather may not propagate cancellation correctly. In general, you should not suppress CancelledError inside coroutines unless you have a specific cleanup pattern that re-raises it after finishing cleanup.

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.

# 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 multiprocessing.

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, remember that the exception objects may contain tracebacks that hold references to frames, 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 is available since Python 3.4, but its behavior with return_exceptions and cancellation has been stable since Python 3.7. In Python 3.10 and later, the event loop policy changed, but gather itself remains the same. If you are using Python 3.11+, you can also use asyncio.TaskGroup for structured concurrency, which provides a more explicit way to manage groups of tasks and handle exceptions. TaskGroup is a newer alternative that may be preferable in new code, but gather remains widely used and compatible with older Python versions.

When migrating from gather to TaskGroup, note that TaskGroup cancels all tasks if any task raises an exception, whereas gather by default propagates the first exception but lets others continue. The choice depends on whether you want fail-fast or partial-success behavior.