Run Multiple Tasks Concurrently in Python

Running one or two async tasks concurrently is straightforward. Running hundreds or thousands requires patterns that manage concurrency, collect results, handle failures, and avoid overwhelming the event loop or the external services you are calling. Python's asyncio module gives you four primary tools for running multiple tasks: gather, TaskGroup, wait, and as_completed.

Each solves a different coordination problem, and choosing the right one determines whether your code is clean and correct or tangled with manual bookkeeping. If you have read about creating and running async tasks, you understand the task lifecycle. This article covers the higher-level patterns that coordinate many tasks at once.

The key difference between these tools is what they optimize for. gather optimizes for simplicity when you want all results in a known order. TaskGroup optimizes for safety by guaranteeing cleanup.

wait optimizes for flexibility when completion order or partial results matter. as_completed optimizes for responsiveness when you want to process results as they arrive.

asyncio.gather: ordered results from concurrent coroutines

gather runs multiple awaitables concurrently and returns a list of results in the same order as the inputs. It is the simplest tool and the right default when you have a known set of coroutines and need all their results.

python

import asyncio
import time
 
async def fetch(url, delay):
    await asyncio.sleep(delay)
    return f"Content of {url}"
 
async def main():
    start = time.perf_counter()
 
    results = await asyncio.gather(
        fetch("https://a.example.com", 0.2),
        fetch("https://b.example.com", 0.1),
        fetch("https://c.example.com", 0.3),
    )
 
    for result in results:
        print(result)
    print(f"Total: {time.perf_counter() - start:.2f}s")
 
asyncio.run(main())

The slowest fetch takes 0.3 seconds, and since all three run concurrently, the total matches that single slowest delay instead of the 0.6 seconds a sequential run of all three would take:

text

Content of https://a.example.com
Content of https://b.example.com
Content of https://c.example.com
Total: 0.30s

All three fetches run concurrently. The total time equals the slowest individual operation, not the sum. Results are returned in the same order as the arguments regardless of which coroutine finishes first.

When one coroutine raises an exception, gather raises it immediately and attempts to cancel the remaining coroutines. To collect exceptions instead, use return_exceptions=True:

python

async def main():
    results = await asyncio.gather(
        safe_fetch("A"),
        failing_fetch("B"),
        safe_fetch("C"),
        return_exceptions=True,
    )
 
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"Task {i} failed: {result}")
        else:
            print(f"Task {i} succeeded: {result}")

This pattern lets all coroutines complete even when some fail. It is essential for batch processing where one failure should not block the rest.

TaskGroup: structured concurrent execution

TaskGroup (Python 3.11+) ensures that all tasks in the group complete before the block exits. If any task fails, all remaining tasks are cancelled, and an ExceptionGroup is raised containing every failure.

python

import asyncio
 
async def worker(name, delay, should_fail=False):
    await asyncio.sleep(delay)
    if should_fail:
        raise ValueError(f"{name} failed")
    return f"{name} done"
 
async def main():
    async with asyncio.TaskGroup() as tg:
        task_a = tg.create_task(worker("A", 0.2))
        task_b = tg.create_task(worker("B", 0.1))
        task_c = tg.create_task(worker("C", 0.3))
 
    print(task_a.result())
    print(task_b.result())
    print(task_c.result())
 
asyncio.run(main())

TaskGroup is preferred over raw create_task() for most concurrent work because it prevents orphaned tasks. A TaskGroup created inside an async with block guarantees that when the block exits, every task in the group has completed. You never need to manually track which tasks you created or ensure they are all awaited.

When combining TaskGroup with a dynamic number of tasks:

python

async def main():
    urls = ["a.example.com", "b.example.com", "c.example.com"]
 
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch(url, 0.1)) for url in urls]
 
    results = [t.result() for t in tasks]
    print(results)

asyncio.wait: flexible completion control

asyncio.wait() gives you finer control over when to stop waiting. It returns two sets: done tasks and pending tasks. The return_when parameter controls the stop condition.

python

import asyncio
 
async def fetch(source, delay):
    await asyncio.sleep(delay)
    return f"Data from {source}"
 
async def main():
    tasks = [
        asyncio.create_task(fetch("API", 0.3)),
        asyncio.create_task(fetch("Cache", 0.1)),
        asyncio.create_task(fetch("DB", 0.2)),
    ]
 
    done, pending = await asyncio.wait(
        tasks, return_when=asyncio.FIRST_COMPLETED
    )
 
    print(f"First done: {done.pop().result()}")
 
    for task in pending:
        task.cancel()
 
asyncio.run(main())

return_when accepts three values:

• ALL_COMPLETED (default): wait until every task finishes.
• FIRST_COMPLETED: return as soon as any task finishes.
• FIRST_EXCEPTION: return as soon as any task raises, or all complete.

wait() is the right tool when you need to race tasks, implement timeouts with partial results, or handle completion dynamically based on which tasks finish first.

asyncio.as_completed: process results as they arrive

as_completed() returns an iterator that yields tasks as they finish, in completion order. This lets you start processing early results while slower tasks are still running.

python

import asyncio
import random
 
async def process_item(item_id):
    delay = random.uniform(0.1, 0.5)
    await asyncio.sleep(delay)
    return f"Processed item {item_id}"
 
async def main():
    coros = [process_item(i) for i in range(10)]
    tasks = [asyncio.create_task(c) for c in coros]
 
    for completed in asyncio.as_completed(tasks):
        result = await completed
        print(result)
 
asyncio.run(main())

as_completed() is ideal for streaming processing where you want to act on results immediately. A web scraper that saves each page as it downloads, a pipeline that processes data chunks as they arrive, or a monitoring system that reports results in real time all benefit from as_completed.

The order of results depends on task completion time, which is unpredictable. If you need to know which input produced which result, store a mapping from task to input:

python

async def main():
    urls = ["a.com", "b.com", "c.com"]
    tasks = {asyncio.create_task(fetch(url, random.uniform(0.1, 0.5))): url
             for url in urls}
 
    for completed in asyncio.as_completed(tasks):
        url = tasks[completed]
        result = await completed
        print(f"{url}: {result}")

Limiting concurrency with Semaphore

When you have thousands of tasks but the resource you are calling can only handle a limited number of concurrent requests, use asyncio.Semaphore to cap concurrency.

python

import asyncio
 
async def fetch_with_limit(sem, url):
    async with sem:
        print(f"Fetching {url}")
        await asyncio.sleep(0.2)
        return f"Content of {url}"

Wrapping the whole 20-URL job in a TaskGroup still guarantees cleanup, while the semaphore inside fetch_with_limit is what actually caps how many of those 20 tasks touch the network at once:

python

async def main():
    urls = [f"https://example.com/page/{i}" for i in range(20)]
    sem = asyncio.Semaphore(5)
 
    async with asyncio.TaskGroup() as tg:
        tasks = [
            tg.create_task(fetch_with_limit(sem, url))
            for url in urls
        ]
 
    for task in tasks:
        print(task.result())
 
asyncio.run(main())

At most 5 fetches run concurrently. When a sixth coroutine reaches async with sem, it blocks until one of the first five exits its async with sem block. The semaphore acts as a gate, ensuring the external service never receives more than 5 concurrent requests.

A semaphore initialized to N allows N concurrent holders. You can think of it as a counter that starts at N, decrements on entry, and increments on exit. When the counter is zero, the next entrant blocks.

Combining patterns: semaphore with as_completed

For large-scale batch processing, combine a semaphore with as_completed:

python

import asyncio
 
async def process(sem, item_id):
    async with sem:
        await asyncio.sleep(0.1)
    return f"Result {item_id}"

Creating all 100 tasks up front does not mean all 100 run at once. Each task still has to pass through the semaphore in process, so only 10 are ever doing real work while the rest wait their turn:

python

async def main():
    items = list(range(100))
    sem = asyncio.Semaphore(10)
 
    tasks = [
        asyncio.create_task(process(sem, item_id))
        for item_id in items
    ]
 
    results = []
    for completed in asyncio.as_completed(tasks):
        result = await completed
        results.append(result)
        print(f"Completed: {len(results)}/100")
 
    print("All done.")
 
asyncio.run(main())

This pattern limits concurrency to 10 while processing results as they complete. It scales from 100 items to 100,000 without modification and without overwhelming resources.

Choosing the right tool

For most batch processing, start with gather() and switch to TaskGroup when you need the safety guarantee. Add a semaphore when you need to limit concurrency. Use as_completed() when you want to stream results rather than wait for all of them.

The next articles in this section cover async iterators and generators for streaming data through async pipelines, and async context managers for managing async resources with deterministic setup and teardown.