Running Coroutines Concurrently with `asyncio.gather()`

Hello! Let's dive into our next lesson on asyncio.

Introduction

In our previous lesson, we unlocked true concurrency by using asyncio.create_task() to schedule coroutines on the event loop. We saw how this allows the event loop to switch between tasks during I/O waits, drastically reducing total execution time. However, the pattern of creating tasks one by one and then awaiting them individually can become cumbersome, especially when dealing with a large number of concurrent operations.

This lesson introduces a more elegant and powerful solution. We will focus on the learning outcome: Use asyncio.gather() to run multiple coroutines concurrently and collect their results. You'll learn how asyncio.gather() simplifies the management of concurrent tasks, making your code more concise and readable without sacrificing performance.

1. From Manual Task Management to asyncio.gather()

While creating and awaiting tasks manually works perfectly well, asyncio provides higher-level APIs to streamline this common pattern. asyncio.gather() is the most fundamental of these APIs. It's designed to run a collection of awaitable objects (like coroutines or tasks) concurrently and "gather" their results into a list.

The key idea is to treat a group of concurrent operations as a single awaitable unit.

To start, let's read a short article that provides a clear example and a helpful analogy for what gather() does.

2. How gather() Works

At first glance, gather() might seem like a simple loop, but it's more sophisticated. When you pass coroutines to asyncio.gather(), it automatically wraps each one in a Task and schedules them on the event loop. It then returns a single Future object that represents the entire group of operations.

When you await this Future, the event loop proceeds with running the tasks concurrently. Your main coroutine remains suspended until all the tasks passed to gather() have completed. Upon completion, the await expression resolves to a list containing the return values of each coroutine, in the same order they were passed in.

A very common and practical pattern is to prepare a list of coroutines and then use the iterable unpacking operator (*) to pass them as separate arguments to gather().

Let's explore these mechanics in more detail.

3. Your Turn: Refactoring to asyncio.gather()

Now, let's apply this knowledge. Below is the solution from our previous lesson, where we used asyncio.create_task() to run two coroutines concurrently. Your task is to refactor the main function to use asyncio.gather() instead.

Your Code to Modify:

import asyncio
import time

async def fetch_data(data_id: int, delay: float):
    """A coroutine that simulates a network request."""
    print(f"Starting to fetch data {data_id}...")
    await asyncio.sleep(delay)
    print(f"Finished fetching data {data_id}.")
    return {"data_id": data_id, "content": f"Data for {data_id}"}

async def main():
    """The main entry point for our program."""
    start_time = time.perf_counter()
    
    print("--- Concurrent Execution with create_task ---")
    # Create tasks and schedule them on the event loop
    task1 = asyncio.create_task(fetch_data(1, 2))
    task2 = asyncio.create_task(fetch_data(2, 3))
    
    # Await the completion of the tasks to get their results
    result1 = await task1
    result2 = await task2
    
    print(f"Result 1: {result1}")
    print(f"Result 2: {result2}")
    
    end_time = time.perf_counter()
    print(f"Total time: {end_time - start_time:.2f} seconds")

if __name__ == "__main__":
    asyncio.run(main())

Instructions:

1. Modify the main function to use asyncio.gather().
2. You should be able to achieve the same result in a single await statement.
3. The results from gather() will be a list. Unpack this list to print the individual results.
4. Run the code and verify that the total time is still approximately 3 seconds.

import asyncio
import time

async def fetch_data(data_id: int, delay: float):
    """A coroutine that simulates a network request."""
    print(f"Starting to fetch data {data_id}...")
    await asyncio.sleep(delay)
    print(f"Finished fetching data {data_id}.")
    return {"data_id": data_id, "content": f"Data for {data_id}"}

async def main():
    """The main entry point for our program."""
    start_time = time.perf_counter()
    
    print("--- Concurrent Execution with gather ---")
    
    # Run coroutines concurrently and gather results
    results = await asyncio.gather(
        fetch_data(1, 2),
        fetch_data(2, 3)
    )
    
    # The results are returned as a list in the order the coroutines were provided
    result1, result2 = results
    
    print(f"Result 1: {result1}")
    print(f"Result 2: {result2}")
    
    end_time = time.perf_counter()
    print(f"Total time: {end_time - start_time:.2f} seconds")

if __name__ == "__main__":
    asyncio.run(main())

4. Error Handling and Advanced Usage

A crucial aspect of any robust system is how it handles failures. The default behavior of asyncio.gather() is to "fail fast." If any of the awaitables passed to it raises an exception, the gather() call is immediately cancelled, and it propagates the first exception it encounters. This can leave other, longer-running tasks in an indeterminate state—they are not automatically cancelled.

For scenarios where you want to ensure all tasks run to completion, even if some fail, gather() provides the return_exceptions=True argument. When set, gather() will treat exceptions as successful results. The returned list will contain the exception objects for the failed tasks alongside the regular return values for the successful ones. This allows your code to process all outcomes without crashing.

Another important point of comparison is with asyncio.TaskGroup (introduced in Python 3.11), which offers more structured concurrency. A TaskGroup guarantees that if one task fails, all other tasks within the group are automatically cancelled.

The following video explains these trade-offs clearly.

To summarize the trade-offs:

• asyncio.gather() (default): Fails fast. Good for tightly coupled operations where one failure means the whole batch is invalid.
• asyncio.gather(..., return_exceptions=True): Runs all tasks to completion. Ideal for independent operations where you want to process as many successful results as possible (e.g., scraping a list of URLs).
• asyncio.TaskGroup: "All or nothing" with guaranteed cleanup. If one task fails, all others in the group are cancelled. This is often the safest and most robust choice for modern Python applications.

Conclusion

In this lesson, you've added asyncio.gather() to your toolkit, a powerful and convenient function for managing concurrent operations.

Key Takeaways:

• asyncio.gather() simplifies running multiple awaitables concurrently by wrapping them in a single Future.
• It collects results in a list, ordered according to the input awaitables.
• It's common to use the * operator to unpack a list of coroutines into gather().
• Error handling is a critical consideration: by default, gather() fails fast, but return_exceptions=True allows you to collect exceptions as results.
• For structured concurrency with guaranteed cleanup, asyncio.TaskGroup is often a more robust alternative.

Preview of the Next Lesson:

We now have two solid patterns for concurrency: manual asyncio.create_task() and the convenient asyncio.gather(). In the next lesson, we will apply these concepts to a practical challenge: writing a program to concurrently fetch data from multiple simulated web endpoints and formally comparing its performance against a sequential approach. This will solidify your understanding of the real-world benefits of asyncio.