How to Use asyncio.gather for Running Tasks in Parallel

Here is the article you requested.

How to Use asyncio.gather for Running Tasks in Parallel

If you have ever felt like your Python program is wasting time just sitting there waiting for something to finish—a file to load, an API to respond, or a database query to return—you are not alone. A lot of us start writing synchronous code, and it works, until it doesn't. That is where asyncio.gather comes in. It lets you run multiple asynchronous tasks at the same time, making your script a whole lot faster.

Think of it this way: instead of cooking one dish at a time, you put the rice on the stove, start chopping vegetables, and set the oven to preheat—all while the rice is cooking. That is concurrency, and asyncio.gather is the kitchen timer that tells you when everything is ready.

What Exactly Is asyncio.gather?

You use asyncio.gather when you have several awaitable objects (like coroutines or tasks) and you want them to run concurrently. This is different from running them one after another. When you write await task_one() and then await task_two(), your program waits for the first one to completely finish before even starting the second one. With gather, you basically say: "Go ahead and start all of these. I will wait for all of them to finish, but they can do their waiting together."

This is fantastic for I/O bound operations. Things like downloading data from three different websites, reading files from disk, or hitting multiple database endpoints.

A Very Simple Example

Let us start with a basic analogy. Imagine you have two functions that simulate some work using asyncio.sleep. This is not real work, but it shows the time difference perfectly.

import asyncio
import time

async def fetch_data_one():
    print("Starting fetch 1")
    await asyncio.sleep(3)
    print("Done with fetch 1")
    return "Data 1"

async def fetch_data_two():
    print("Starting fetch 2")
    await asyncio.sleep(2)
    print("Done with fetch 2")
    return "Data 2"

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

    # The sequential way
    result1 = await fetch_data_one()
    result2 = await fetch_data_two()

    end = time.time()
    print(f"Sequential took {end - start:.2f} seconds")
    print(result1, result2)

asyncio.run(main())

If you run this, you will see it takes about 5 seconds. That is 3 + 2. Because the second one does not start until the first one is done. That is slow. It is not using your computer's time well at all.

Now, let us use gather.

import asyncio
import time

async def fetch_data_one():
    print("Starting fetch 1")
    await asyncio.sleep(3)
    print("Done with fetch 1")
    return "Data 1"

async def fetch_data_two():
    print("Starting fetch 2")
    await asyncio.sleep(2)
    print("Done with fetch 2")
    return "Data 2"

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

    # The concurrent way
    results = await asyncio.gather(
        fetch_data_one(),
        fetch_data_two()
    )

    end = time.time()
    print(f"Concurrent took {end - start:.2f} seconds")
    print(results)

asyncio.run(main())

Now the output looks like this:

Starting fetch 1
Starting fetch 2
Done with fetch 2
Done with fetch 1
Concurrent took 3.00 seconds

Notice the time dropped from 5 seconds to about 3 seconds. That is the power of concurrency. The longest task (3 seconds) dictated the total time. The 2-second task finished while the 3-second task was still going.

Handling the Results

Look at how we got the results. asyncio.gather returns a list of results, in the exact same order you passed the tasks. So if you pass task_a() first and task_b() second, the first item in the results list will be the return value of task_a(), even if task_b() finished first. This is a very useful guarantee from PythonSkillset.

results = await asyncio.gather(task_a(), task_b())
result_a = results[0]
result_b = results[1]

What Happens When One Task Fails?

This is a crucial point for any real world script. By default, if any one of the tasks in gather raises an exception, gather will raise that exception immediately. But here is the catch—it does not wait for the other tasks to finish. The other tasks are not cancelled automatically in Python 3.8 or earlier, but in Python 3.9+, if you do not handle the exception, the other tasks might be cancelled depending on your event loop policy.

To avoid surprises, you can use return_exceptions=True.

results = await asyncio.gather(
    risky_task(),
    safe_task(),
    return_exceptions=True
)

With this setting, if risky_task fails, gather does not raise an error. Instead, the exception object itself is placed in the results list for that position. You then have to check each result to see if it is an exception or a normal return value.

for i, res in enumerate(results):
    if isinstance(res, Exception):
        print(f"Task {i} failed with {res}")
    else:
        print(f"Task {i} returned {res}")

This is the pattern I use most often at PythonSkillset because it gives you fine control over error recovery.

When Should You Use It?

You should use asyncio.gather whenever you have a known set of independent I/O operations. "Independent" is the key word. If task 2 needs the result of task 1, you cannot use gather. You have to do them one after the other.

Good examples of independent tasks: - Fetching data from three different APIs. - Reading several configuration files at startup. - Sending emails to a list of users. - Running multiple database queries that do not depend on each other.

Bad examples: - Task 2 needs the user ID returned by Task 1. - Writing to a file should happen after reading it.

A Real World Example from PythonSkillset

Let me show you how I use it in a real script for checking server health. Imagine we have a list of URLs and we need to check if they respond with a 200 status code.

import asyncio
import aiohttp

async def check_server(session, url):
    try:
        async with session.get(url, timeout=5) as response:
            return url, response.status
    except Exception as e:
        return url, str(e)

async def health_check(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [check_server(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results

urls_to_check = [
    "https://pythonskillset.com",
    "https://example.com",
    "https://nonexistent.server.local"
]

results = asyncio.run(health_check(urls_to_check))
for url, status in results:
    print(f"{url}: {status}")

This script will hit all three URLs at the same time. The total time is roughly the time of the slowest server, not the sum of all three. You can see how this saves massive time in real monitoring scripts.

A Small Gotcha: Unpacking the Tasks List

Notice in the health check example we used *tasks. The gather function takes arguments as separate positional arguments, not a single list. If you have a list of coroutines, you need to unpack it with the * operator. Forgetting this is a common mistake.

# Wrong
await asyncio.gather(tasks)

# Right
await asyncio.gather(*tasks)

Wrapping Up

asyncio.gather is one of those tools that once you start using, you will wonder how you lived without it. It turns a slow, sequential script into a fast concurrent one with very little code change. The main things to remember are:

• Use it for independent I/O tasks.
• Pass return_exceptions=True if you want to handle failures gracefully.
• Unpack your task list with *.
• The results come back in the order you passed the tasks.

So the next time you are writing code that waits for one thing, then another, then another, stop and think: "Can I use gather here?" The answer is probably yes, and your users (and your sleep schedule) will thank you.