asyncio.gather(): Run Concurrent Tasks

asyncio.gather() is the high-level, convenient function for running multiple coroutines concurrently and waiting for all of them to complete. It collects their results into a single list, maintains their order, and provides clean error-handling options. This function transforms verbose task-creation code into a single elegant call—essential for I/O-bound workloads like API requests, database queries, and file operations that benefit from concurrent execution.

Key Takeaways

• asyncio.gather() runs multiple coroutines concurrently on the same event loop, returning a list of results in the same order as input
• Total execution time equals the longest-running task, not the sum of all tasks (true concurrency, not sequencing)
• return_exceptions=True handles failures gracefully by returning exceptions as list items instead of raising them, allowing partial success
• Better than manual task creation when you have a group of coroutines to run together; use create_task() when you need independent background tasks

---

How Is asyncio.gather() Better Than Manual Task Creation?

Without gather(), running multiple coroutines concurrently requires creating each as a task and awaiting them individually:

async def main_manual():
    task1 = asyncio.create_task(fetch_api("users", 1))
    task2 = asyncio.create_task(fetch_api("products", 2))
    task3 = asyncio.create_task(fetch_api("orders", 1.5))

    # Await each one individually—verbose
    result1 = await task1
    result2 = await task2
    result3 = await task3

    print(result1, result2, result3)

This pattern is repetitive for 3 tasks and becomes unwieldy for 10 or 100. asyncio.gather() simplifies by accepting any number of awaitable objects (coroutines or tasks) and returning a single list of results:

async def main_with_gather():
    results = await asyncio.gather(
        fetch_api("users", 1),
        fetch_api("products", 2),
        fetch_api("orders", 1.5)
    )
    print(results)  # Single call, clean syntax

Both run the tasks concurrently, but gather() is more concise and maintains result order automatically.

---

How Do You Use asyncio.gather() with Multiple Coroutines?

gather() takes one or more coroutines (or awaitable objects) as positional arguments, runs them concurrently on the event loop, and returns a list of their results.

import asyncio
import time

async def fetch_from_api(endpoint: str, delay: float):
    """Simulates a slow API request."""
    print(f"[{time.time():.2f}] Fetching from {endpoint}...")
    await asyncio.sleep(delay)
    result = {"endpoint": endpoint, "status": "success"}
    print(f"[{time.time():.2f}] Finished fetching from {endpoint}")
    return result

async def main():
    start = time.time()

    # Run three coroutines concurrently
    results = await asyncio.gather(
        fetch_from_api("users", 1),
        fetch_from_api("products", 2),
        fetch_from_api("orders", 1.5)
    )

    elapsed = time.time() - start
    print(f"\nCompleted in {elapsed:.2f} seconds")
    print(f"Results: {results}")

asyncio.run(main())

Output:

[0.00] Fetching from users...
[0.00] Fetching from products...
[0.00] Fetching from orders...
[1.00] Finished fetching from users
[1.50] Finished fetching from orders
[2.00] Finished fetching from products

Completed in 2.00 seconds
Results: [{'endpoint': 'users', 'status': 'success'}, 
          {'endpoint': 'products', 'status': 'success'}, 
          {'endpoint': 'orders', 'status': 'success'}]

Key observations:

• All three tasks start immediately (concurrent)
• The total time is ~2 seconds, the duration of the longest task (products = 2 sec), not the sum (4.5 sec)
• Results are returned in the same order as the input coroutines
• If any coroutine returns a value, that value appears in the results list at the corresponding index

Gathering Tasks Instead of Coroutines

You can also pass pre-created tasks to gather():

async def main():
    # Create tasks explicitly
    task1 = asyncio.create_task(fetch_from_api("users", 1))
    task2 = asyncio.create_task(fetch_from_api("products", 2))

    # Gather the tasks
    results = await asyncio.gather(task1, task2)
    print(results)

This is useful if you need to interact with tasks before gathering (e.g., cancel them, check their status).

---

How Do You Handle Errors in Concurrent Tasks?

By default, if any task raises an exception, gather() propagates it immediately, and the exception stops awaiting the result. Other tasks continue running in the background but their results are lost.

Default Behavior: Raise on First Exception

async def fetch_or_fail(endpoint: str, delay: float, fail: bool = False):
    """A coroutine that might raise an exception."""
    print(f"Fetching from {endpoint}...")
    await asyncio.sleep(delay)
    if fail:
        raise ValueError(f"Error fetching {endpoint}")
    return {"endpoint": endpoint, "status": "success"}

async def main_default():
    try:
        results = await asyncio.gather(
            fetch_or_fail("users", 1),
            fetch_or_fail("products", 2, fail=True),  # This will fail
            fetch_or_fail("orders", 1.5)
        )
        print(results)
    except ValueError as e:
        print(f"Caught exception: {e}")

asyncio.run(main_default())

Output:

Fetching from users...
Fetching from products...
Fetching from orders...
Caught exception: Error fetching products

Graceful Handling: return_exceptions=True

To continue processing even if some tasks fail, use return_exceptions=True. Exceptions are returned as list items instead of being raised:

async def main_with_exceptions():
    results = await asyncio.gather(
        fetch_or_fail("users", 1),
        fetch_or_fail("products", 2, fail=True),
        fetch_or_fail("orders", 1.5),
        return_exceptions=True  # Catch exceptions, don't raise them
    )

    print("\n--- Results ---")
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"Task {i} failed: {result}")
        else:
            print(f"Task {i} succeeded: {result}")

asyncio.run(main_with_exceptions())

Output:

--- Results ---
Task 0 succeeded: {'endpoint': 'users', 'status': 'success'}
Task 1 failed: Error fetching products
Task 2 succeeded: {'endpoint': 'orders', 'status': 'success'}

With return_exceptions=True, you get partial results: successful tasks return their values, failed tasks return their exception objects. This is essential for resilient concurrent applications where some failures are acceptable.

---

How Do You Gather a Dynamic List of Coroutines?

Often, the number of tasks is determined at runtime. You can build a list of coroutines and unpack it into gather() using the * operator:

async def fetch_data(url: str):
    """Fetch data from a URL."""
    await asyncio.sleep(0.5)  # Simulate network delay
    return {"url": url, "data": "sample"}

async def main():
    urls = ["https://api.example.com/1", "https://api.example.com/2", "https://api.example.com/3"]

    # Build a list of coroutines
    coroutines = [fetch_data(url) for url in urls]

    # Unpack the list into gather()
    results = await asyncio.gather(*coroutines, return_exceptions=True)

    for result in results:
        print(result)

asyncio.run(main())

This pattern is essential for processing variable numbers of items concurrently (e.g., a list of user IDs, file paths, or API endpoints).

---

When Should You Use gather() vs. create_task()?

Use Case	Function	Reason
Run multiple tasks now, wait for all	asyncio.gather()	High-level, returns results directly
Start background task, continue execution	asyncio.create_task()	Immediate return; task runs independently
Known set of coroutines to await	asyncio.gather()	Cleaner syntax, automatic result collection
Spawn independent, long-running tasks	asyncio.create_task()	More flexible; tasks live beyond the current function
Need to cancel/check status of individual tasks	asyncio.create_task()	Direct task object reference required

# Use gather() for coordinated waiting
async def parallel_api_calls():
    results = await asyncio.gather(
        fetch_api("endpoint1"),
        fetch_api("endpoint2"),
        fetch_api("endpoint3")
    )
    return results

# Use create_task() for independent background work
async def background_logging():
    task = asyncio.create_task(log_periodically())
    # Continue doing other things
    return task

---

What Are Best Practices for Using asyncio.gather()?

Best Practices:

• Always use return_exceptions=True in production unless you expect all tasks to succeed and want to fail fast
• Pass coroutines, not pre-awaited values: gather(coro1(), coro2()) not gather(await coro1(), await coro2())
• Limit concurrency for large numbers of tasks using asyncio.Semaphore() to avoid overwhelming resources
• Document task semantics: clarify whether all tasks must succeed, whether order matters, and how timeouts are handled

Anti-Patterns to Avoid:

• Don't use gather() for sequential work: If you need task B to start after task A finishes, don't use gather() (use explicit awaits instead)
• Don't ignore exceptions with return_exceptions=True without checking for them in results
• Don't create thousands of unbounded tasks: Always use a concurrency limit (Semaphore) for large workloads

# Anti-pattern: unlimited concurrency with large task count
async def main_bad():
    tasks = [fetch_api(url) for url in 10000 urls]
    results = await asyncio.gather(*tasks)  # Creates 10k tasks at once

# Better: limit concurrency
async def main_good():
    semaphore = asyncio.Semaphore(10)  # Max 10 concurrent requests
    
    async def bounded_fetch(url):
        async with semaphore:
            return await fetch_api(url)
    
    tasks = [bounded_fetch(url) for url in urls]
    results = await asyncio.gather(*tasks, return_exceptions=True)

---

Frequently Asked Questions

Does gather() preserve the order of results?

Yes, always. Results are returned in the same order as the input coroutines, regardless of which task finishes first.

What happens if you pass an empty list to gather()?

gather() returns an empty list immediately. It does not raise an error.

results = await asyncio.gather()  # Returns []

Can you pass a mix of coroutines and tasks to gather()?

Yes, gather() accepts any awaitable objects: coroutines, tasks, futures, etc.

task = asyncio.create_task(fetch_api("users"))
coro = fetch_api("products")
results = await asyncio.gather(task, coro)

How do you implement a timeout for gather()?

Use asyncio.wait_for() to wrap the gather() call:

try:
    results = await asyncio.wait_for(asyncio.gather(*tasks), timeout=5.0)
except asyncio.TimeoutError:
    print("Gather timed out after 5 seconds")

Should you use gather() for very large numbers of tasks?

For thousands of tasks, use asyncio.Semaphore() to limit concurrency. Unbounded concurrency can exhaust system resources (open connections, memory). A semaphore queues excess tasks, running them as earlier tasks complete.

---

Further Reading

• Python Official asyncio.gather() Documentation
• Real Python: Async IO in Python: A Complete Walkthrough
• asyncio Patterns and Best Practices