Python gather multiple coroutines with asyncio.gather

When you need to run several coroutines concurrently in Python, asyncio.gather is the standard tool for the job. The function accepts one or more awaitables, schedules them on the current event loop, and returns a single coroutine that awaits all of them. The core pattern for python gather multiple coroutines looks like this:

import asyncio

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

async def main():
    results = await asyncio.gather(
        fetch_data("first", 0.5),
        fetch_data("second", 1.0),
        fetch_data("third", 0.2),
    )
    print(results)

asyncio.run(main())

The coroutines passed to gather start executing as soon as the event loop gets a chance to run them, not sequentially one after another. The await on the result suspends main until every coroutine completes, and the printed list contains all three return values.

The basic gather pattern

asyncio.gather accepts any number of awaitables as positional arguments. Each argument can be a coroutine object or an already-created task. The function returns a future that resolves to a list of results, one entry per input, in the same order as the inputs were passed.

The await is what blocks the calling coroutine until the whole batch finishes. Without it, gather only schedules the work; the returned future must be awaited or passed to the event loop explicitly for the coroutines to run.

How gather schedules the coroutines

When you call asyncio.gather, it wraps each input in a task internally. That means the coroutines are scheduled on the event loop and can interleave at each await point. The order in which they finish is not guaranteed, but the order of the returned list is.

Because the scheduling happens through the event loop, the concurrency is cooperative. Each coroutine runs until it hits an await, then control returns to the loop, which can advance another coroutine. This is why asyncio.sleep in the example produces overlapping execution: the three calls do not run one after the other; they progress together.

Result ordering

The list returned by gather preserves the order of the inputs, not the order of completion. This is a useful property when you need to correlate results with the original calls.

async def main():
    results = await asyncio.gather(
        fetch_data("slow", 1.0),
        fetch_data("fast", 0.1),
    )
    # results[0] corresponds to "slow", results[1] to "fast"

The slow coroutine finishes last, yet its result still appears at index 0. If you need results in completion order, asyncio.as_completed is the better API, because it yields each result as soon as it is ready.

Handling exceptions with return_exceptions

By default, if any coroutine raises an exception, gather propagates that exception immediately. The remaining coroutines are not cancelled; they continue running, but their results are discarded from the caller's perspective. To collect exceptions as results instead, set return_exceptions=True.

async def fail() -> str:
    raise RuntimeError("boom")

async def main():
    results = await asyncio.gather(
        fetch_data("ok", 0.2),
        fail(),
        return_exceptions=True,
    )
    for item in results:
        if isinstance(item, Exception):
            print("failed:", item)
        else:
            print("ok:", item)

With return_exceptions=True, the exception object is placed in the result list at the position of the failed coroutine, and gather does not raise. This is useful when you want to run a batch of independent operations and handle failures per item rather than aborting the whole batch.

Cancellation behavior

If the task awaiting gather is cancelled, gather cancels all the child tasks it created. A child task that receives a CancelledError will propagate it, and the cancellation propagates up through the await. If return_exceptions=True, the CancelledError is captured as an exception object in the result list instead of propagating.

This matters in production code where a request timeout or a shutdown signal may cancel the awaiting task. The child coroutines get a chance to run their finally blocks and clean up resources, but they cannot prevent the cancellation from completing.

gather vs create_task, wait, and as_completed

The choice depends on what you need from the batch. gather is the most direct when you want a single await point and a complete result list. create_task is appropriate when you need to store tasks and await them later, or when you want to attach callbacks. wait gives you explicit control over timeouts and lets you inspect which tasks are still pending. as_completed is the right tool when the first finished result should be handled immediately.

When gather is the right choice

Use gather when you have a fixed set of coroutines, you want all results collected in input order, and a single await point is enough. That covers most batch operations: fetching several resources, running independent validation steps, or waiting for multiple background jobs to finish before continuing.

Do not use gather when you need incremental results, because the list is only available once everything completes. Do not use it when you need a timeout on the whole operation, because gather has no timeout parameter; asyncio.wait with a timeout argument is more direct for that case.

Concurrency and performance considerations

The concurrency gain comes from awaiting inside the coroutines, not from gather itself. gather does not make blocking code non-blocking. If a coroutine calls a blocking function such as time.sleep or a synchronous database driver, it will block the entire event loop, and the other coroutines will not make progress until it returns. For blocking work, use asyncio.to_thread or a dedicated thread pool.

async def main():
    results = await asyncio.gather(
        asyncio.to_thread(blocking_io_call, "first"),
        asyncio.to_thread(blocking_io_call, "second"),
    )

Another constraint: pass each coroutine to exactly one gather call. If the same coroutine object is passed to two gather calls, the second call will raise an error because the coroutine is already scheduled. If you need to share work across multiple consumers, create the task once with asyncio.create_task and pass the task object to each gather call.