asyncio Tasks, gather() and Concurrent Execution

Sequential await calls run one after the other — each completes before the next starts. For making multiple independent I/O calls (fetching data from several APIs, running several database queries), running them concurrently reduces total time from the sum of all durations to approximately the duration of the slowest one. asyncio.gather() and asyncio.create_task() are the two primary tools for concurrent execution. Understanding when and how to use them is one of the most impactful async patterns for FastAPI — particularly for route handlers that need data from multiple sources.

asyncio.gather() — Run Coroutines Concurrently #

import asyncio
import httpx

async def fetch(client: httpx.AsyncClient, url: str) -> dict:
    response = await client.get(url, timeout=10.0)
    response.raise_for_status()
    return response.json()

async def main():
    base = "https://jsonplaceholder.typicode.com"

    async with httpx.AsyncClient() as client:
        # ── Sequential — total ≈ sum of each request time ──────────────────────
        user  = await fetch(client, f"{base}/users/1")       # wait
        posts = await fetch(client, f"{base}/posts?userId=1") # then wait
        todos = await fetch(client, f"{base}/todos?userId=1") # then wait
        # Total ≈ 300ms (if each takes 100ms)

        # ── Concurrent — total ≈ slowest single request ────────────────────────
        user, posts, todos = await asyncio.gather(
            fetch(client, f"{base}/users/1"),
            fetch(client, f"{base}/posts?userId=1"),
            fetch(client, f"{base}/todos?userId=1"),
        )
        # Total ≈ 100ms — all three run simultaneously!

asyncio.run(main())

asyncio.create_task() — Background Tasks #

import asyncio

async def slow_operation(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"{name} done after {delay}s"

async def main():
    # create_task() schedules a coroutine to run "in the background"
    # Returns a Task object immediately — the coroutine starts running concurrently
    task1 = asyncio.create_task(slow_operation("A", 1.0))
    task2 = asyncio.create_task(slow_operation("B", 0.5))
    task3 = asyncio.create_task(slow_operation("C", 1.5))

    # Do other work while tasks run in background...
    print("Tasks started — doing other work")
    await asyncio.sleep(0)   # yield to let tasks start

    # Wait for all tasks to complete
    result1 = await task1   # 1.0s from start
    result2 = await task2   # already done (only 0.5s)
    result3 = await task3   # 1.5s from start
    # Total ≈ 1.5s (not 3.0s)

    print(result1, result2, result3)

# ── gather vs create_task ──────────────────────────────────────────────────────
# gather() is simpler when you have all coroutines up front
# create_task() gives more control: cancel, add callbacks, check status

async def with_tasks():
    task = asyncio.create_task(slow_operation("X", 2.0))

    # Check task status
    print(task.done())     # False — still running
    print(task.cancelled())  # False

    # Cancel a task
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Task was cancelled")

asyncio.run(main())

Bounded Concurrency with Semaphore #

import asyncio
import httpx

async def fetch_with_semaphore(
    client: httpx.AsyncClient,
    url: str,
    semaphore: asyncio.Semaphore,
) -> dict:
    async with semaphore:   # blocks if 10 coroutines already running
        response = await client.get(url, timeout=30.0)
        return {"url": url, "status": response.status_code, "data": response.json()}

async def fetch_all_bounded(urls: list[str], max_concurrent: int = 10) -> list[dict]:
    """Fetch all URLs concurrently but with a maximum of max_concurrent at once."""
    semaphore = asyncio.Semaphore(max_concurrent)

    async with httpx.AsyncClient() as client:
        tasks = [
            asyncio.create_task(fetch_with_semaphore(client, url, semaphore))
            for url in urls
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    return [r for r in results if not isinstance(r, Exception)]

# Process 100 URLs, at most 10 at once
urls = [f"https://api.example.com/posts/{i}" for i in range(1, 101)]
results = asyncio.run(fetch_all_bounded(urls, max_concurrent=10))

asyncio.wait() — Fine-Grained Control #

import asyncio

async def main():
    tasks = [
        asyncio.create_task(slow_operation("A", 1.0)),
        asyncio.create_task(slow_operation("B", 0.5)),
        asyncio.create_task(slow_operation("C", 2.0)),
    ]

    # Wait until ALL tasks complete
    done, pending = await asyncio.wait(tasks)
    for task in done:
        print(await task)

    # Wait until FIRST task completes (return_when=FIRST_COMPLETED)
    tasks2 = [asyncio.create_task(slow_operation(f"T{i}", i)) for i in range(1, 4)]
    done, pending = await asyncio.wait(tasks2, return_when=asyncio.FIRST_COMPLETED)
    # Cancel remaining
    for task in pending:
        task.cancel()
    for task in done:
        print(f"First completed: {task.result()}")

Common Mistakes #

Mistake 1 — Gathering without error handling #

❌ Wrong — one failure cancels all:

results = await asyncio.gather(fetch(url1), fetch(url2), fetch(url3))
# If fetch(url2) raises — all three are cancelled and exception propagates

✅ Correct — handle partial failures:

results = await asyncio.gather(fetch(url1), fetch(url2), fetch(url3),
                                return_exceptions=True)
good = [r for r in results if not isinstance(r, Exception)]   # ✓

Mistake 2 — Unbounded concurrency overwhelming an API #

❌ Wrong — 1000 concurrent requests at once:

await asyncio.gather(*[fetch(url) for url in urls])   # 1000 concurrent — rate limited!

✅ Correct — use a semaphore:

sem = asyncio.Semaphore(10)
await asyncio.gather(*[fetch_with_semaphore(url, sem) for url in urls])   # ✓

Mistake 3 — Awaiting tasks after gather — double wait #

❌ Wrong — gather already awaited them:

tasks = [asyncio.create_task(f()) for f in funcs]
results = await asyncio.gather(*tasks)
for task in tasks:
    r = await task   # task is already done — redundant but works for completed tasks

✅ Correct — use gather’s results directly:

results = await asyncio.gather(*[f() for f in funcs])   # ✓ results already available

Quick Reference #