Concurrent Operations with `asyncio.gather`

The Power of Gathering: From Sequential to Concurrent

Imagine a scenario where your application needs to fetch data from three different, independent network endpoints. In a synchronous world, you would make these calls one after another, and the total time would be the sum of all individual call durations.

With asyncio, we can do much better. asyncio.gather allows us to "gather" all these operations and run them concurrently. The event loop executes them all "at the same time," switching between them during their I/O wait periods. The total time is then limited only by the single longest operation, not the sum of all of them.

This image perfectly illustrates the performance gain:

Let's watch a practical demonstration of this. The following video compares fetching multiple items from an API sequentially versus concurrently with gather, measuring the significant speedup.

How to Use asyncio.gather

The basic syntax for gather is straightforward. You pass it a series of awaitable objects (coroutines or tasks).

import asyncio

async def my_coroutine(id, delay):
    await asyncio.sleep(delay)
    return f"Result from {id}"

async def main():



    # gather takes coroutine objects as arguments
    results = await asyncio.gather(
        my_coroutine(1, 1.5),
        my_coroutine(2, 0.5),
        my_coroutine(3, 1.0)
    )
    print(results)

asyncio.run(main())

Expected Output:

['Result from 1', 'Result from 2', 'Result from 3']

There are two critical characteristics of gather to note here:

1. Concurrency: Although my_coroutine(1, 1.5) is the longest, the total execution time will be ~1.5 seconds, not 3.0 seconds (1.5 + 0.5 + 1.0).
2. Order of Results: The results list is ['Result from 1', 'Result from 2', 'Result from 3']. gather always returns the results in the same order that the awaitables were passed to it, regardless of their completion order. This is incredibly convenient for processing results, as you don't need to manually map results back to their original calls.

This article provides a concise reference on the gather function's signature and behavior.

When you pass coroutine objects directly to gather, it automatically schedules them as tasks for you. This is the most common and convenient way to use it. However, if you need more control (e.g., to cancel a specific task before gather completes), you can create tasks manually with asyncio.create_task and pass those Task objects to gather. The end result of awaiting gather is the same.

Building Resilient Systems: Handling Exceptions

In a distributed system, failure is inevitable. A network call might time out, a service might be down, or an API might return an error. How gather behaves in the face of exceptions is critical for building robust applications.

By default, gather has a "fail-fast" behavior. If any of the awaitables passed to it raises an exception, gather immediately cancels the waiting and propagates that first exception. You lose the results of all other awaitables, even those that completed successfully.

Consider this code:

import asyncio

async def success_coro():
    await asyncio.sleep(1)
    return "Success"

async def fail_coro():
    await asyncio.sleep(0.5)
    raise ValueError("Something went wrong")

async def main():
    try:
        results = await asyncio.gather(
            success_coro(),
            fail_coro()
        )
        print(results)
    except ValueError as e:
        print(f"Caught an exception: {e}")



        # The result from success_coro() is lost!

asyncio.run(main())

Output:

Caught an exception: Something went wrong

This is often not what you want in a microservices context. If you're fetching data for a user dashboard from three services and one fails, you'd ideally want to show the data from the two successful services and an error message for the failed one, rather than showing a full-page error.

The return_exceptions=True Pattern

To handle this, gather provides a crucial parameter: return_exceptions. When set to True, gather will not raise an exception. Instead, it will wait for all awaitables to finish and include any exception objects in the results list, right alongside the successful return values.

This pattern is fundamental for building resilient systems. It allows you to process partial successes and failures gracefully.

The video from Corey Schafer also provides an excellent discussion on gather versus a newer alternative, TaskGroup, highlighting their differences in error handling. While TaskGroup enforces "all or nothing" semantics (which we covered in the last lesson), gather with return_exceptions=True is the go-to tool for "best effort" scenarios.

Practical Exercise: The Aggregating API Gateway

Let's put this into practice. You are building an API gateway that serves a user's dashboard. To do this, it needs to concurrently fetch three pieces of information from different downstream microservices:

1. The user's profile.
2. The user's recent orders.
3. The user's notification settings.

One of these services might fail. Your gateway should still return as much data as possible.

Your Task:

1. Write three coroutines: fetch_profile(user_id), fetch_orders(user_id), and fetch_notifications(user_id).
2. Each coroutine should simulate a network call with asyncio.sleep() using different delays.
3. Make one of the coroutines, for instance fetch_orders, raise a TimeoutError("Order service timed out").
4. In your main function, use asyncio.gather with return_exceptions=True to call these three coroutines for a given user_id.
5. Process the list of results from gather. Construct a final JSON-like dictionary representing the aggregated response.
   • If a result is a success, add it to the dictionary.
   • If a result is an exception, add an error message for that key.

Example final output dictionary:

{
    "profile": {"name": "Test User", "email": "test@example.com"},
    "orders": {"error": "Order service timed out"},
    "notifications": {"email_enabled": True, "sms_enabled": False}
}

import asyncio
import time

async def fetch_profile(user_id: int) -> dict:
    """Fetches user profile data."""
    print(f"Fetching profile for user {user_id}...")
    await asyncio.sleep(0.8)
    return {"name": "Test User", "email": "test@example.com"}

async def fetch_orders(user_id: int) -> list:
    """Fetches user orders. This one will fail."""
    print(f"Fetching orders for user {user_id}...")
    await asyncio.sleep(1.2)
    raise TimeoutError("Order service timed out")

async def fetch_notifications(user_id: int) -> dict:
    """Fetches user notification settings."""
    print(f"Fetching notifications for user {user_id}...")
    await asyncio.sleep(0.5)
    return {"email_enabled": True, "sms_enabled": False}

async def get_user_dashboard_data(user_id: int) -> dict:
    """Aggregates data for the user dashboard using asyncio.gather."""
    
    start_time = time.perf_counter()

    profile_coro = fetch_profile(user_id)
    orders_coro = fetch_orders(user_id)
    notifications_coro = fetch_notifications(user_id)




    # Run all coroutines concurrently and collect results/exceptions
    results = await asyncio.gather(
        profile_coro,
        orders_coro,
        notifications_coro,
        return_exceptions=True
    )
    
    end_time = time.perf_counter()
    print(f"\nData aggregation took {end_time - start_time:.2f} seconds.")




    # Process results to build the final response
    dashboard_data = {}
    



    # Profile data
    if isinstance(results[0], Exception):
        dashboard_data["profile"] = {"error": str(results[0])}
    else:
        dashboard_data["profile"] = results[0]




    # Orders data
    if isinstance(results[1], Exception):
        dashboard_data["orders"] = {"error": str(results[1])}
    else:
        dashboard_data["orders"] = results[1]




    # Notifications data
    if isinstance(results[2], Exception):
        dashboard_data["notifications"] = {"error": str(results[2])}
    else:
        dashboard_data["notifications"] = results[2]

    return dashboard_data

async def main():
    user_id = 123
    print(f"--- Building dashboard for user {user_id} ---")
    dashboard = await get_user_dashboard_data(user_id)
    print("\n--- Aggregated Dashboard Response ---")
    import json
    print(json.dumps(dashboard, indent=2))


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

Conclusion

You've now taken a deep dive into asyncio.gather, one of the most practical and frequently used tools in the asyncio toolkit. You've seen how it enables massive performance gains for I/O-bound work and, most importantly, how to use it to build resilient systems that can gracefully handle the partial failures common in microservice architectures.

Key Takeaways:

• asyncio.gather is your primary tool for running a collection of independent awaitables concurrently.
• It significantly reduces total execution time for I/O-bound operations by running them in parallel, limited only by the longest-running operation.
• It conveniently returns results in the same order as the inputs, simplifying data processing.
• Using return_exceptions=True is the standard pattern for building robust services that can tolerate and report on partial failures, rather than failing completely.

Preview of the Next Lesson:

So far, our entire discussion has centered on I/O-bound tasks—operations like network requests or database queries where the program spends most of its time waiting. But what about tasks that are CPU-bound, like complex calculations, data compression, or image processing? Running these in an async function would block the entire event loop, defeating the purpose of asyncio.

In the next lesson, we will distinguish between I/O-bound and CPU-bound tasks and explore the correct concurrency models for each, introducing Python's multiprocessing as the solution for true parallelism with CPU-heavy workloads.