Concurrent Requests with asyncio.gather in FastAPI

Key takeaways:

  • A fan-out costs the slowest upstream, not the sum of all of them — but only if you never await them one at a time.
  • Plain asyncio.gather throws away good results the moment one branch fails; return_exceptions=True keeps them.
  • asyncio.TaskGroup cancels siblings on failure and raises an ExceptionGroup, which you handle with except*.
  • asyncio.timeout gives the endpoint a latency budget; a per-call budget lets it degrade instead of failing.
  • A semaphore converts an unbounded connection spike into controlled queuing.

This page is the constructive half of Async Correctness and Concurrency. Once you have followed Running Sync Code in a Threadpool and the loop is genuinely free, this is what you do with that freedom.

Serial awaits versus a concurrent fan-out Awaiting three upstreams of 300, 200 and 100 milliseconds one after another takes 600 milliseconds. Issuing them together takes 300 milliseconds, the duration of the slowest one. Three upstreams: 300ms, 200ms, 100ms Serial awaits profile 300ms orders 200ms prefs = 0.6s Concurrent fan-out profile 300ms orders 200ms prefs = 0.3s Both timings below are measured by the endpoint itself, not estimated.
A fan-out is bounded by its slowest branch. Every serial await you remove is latency the caller stops paying.

The Problem This Solves

An endpoint assembles a response from three or four internal services: a profile service, an order history, a preferences store. Written naively it awaits each in turn, and the endpoint's latency becomes the sum of theirs. The loop is not blocked — every await yields correctly — but nothing overlaps, because nothing was ever started concurrently.

The fix is trivially small and the failure modes are not. Fanning out changes what happens when one upstream is down, when one is slow, and when a traffic spike multiplies your fan-out factor by your request rate and lands on a service that was sized for something gentler. This guide covers all four.

Why It Happens: await Starts and Waits

await coro does two things at once — it schedules the coroutine and it suspends until the result is ready. Because those are fused, a sequence of awaits can never overlap. To overlap, you must first turn the coroutines into scheduled tasks and only then wait for them, which is exactly what asyncio.gather and asyncio.TaskGroup do.

The measured baseline, from an app whose three "upstreams" sleep 300 ms, 200 ms and 100 ms:

@app.get("/serial")
async def serial() -> dict:
    """The baseline nobody wants: awaiting each call one after another."""
    started = time.perf_counter()
    results = [await call_upstream(n, d) for n, d in UPSTREAMS.items()]
    return {"elapsed_s": round(time.perf_counter() - started, 2), "results": results}


@app.get("/gather")
async def gather() -> dict:
    """The same three calls issued concurrently: wall clock collapses to the slowest one."""
    started = time.perf_counter()
    results = await asyncio.gather(*[call_upstream(n, d) for n, d in UPSTREAMS.items()])
    return {
        "elapsed_s": round(time.perf_counter() - started, 2),
        "slowest_upstream_s": max(UPSTREAMS.values()),
        "results": results,
    }

Real output from the verification harness:

$ GET /serial
200 OK
{
  "elapsed_s": 0.6,
  "results": [
    {
      "upstream": "profile",
      "took_s": "0.3"
    },
    {
      "upstream": "orders",
      "took_s": "0.2"
    },
    {
      "upstream": "prefs",
      "took_s": "0.1"
    }
  ]
}

$ GET /gather
200 OK
{
  "elapsed_s": 0.3,
  "slowest_upstream_s": 0.3,
  "results": [
    {
      "upstream": "profile",
      "took_s": "0.3"
    },
    {
      "upstream": "orders",
      "took_s": "0.2"
    },
    {
      "upstream": "prefs",
      "took_s": "0.1"
    }
  ]
}

0.6 s becomes 0.3 s, and elapsed_s equals slowest_upstream_s exactly. Note also that gather preserved argument order in its results even though prefs finished first — result ordering is positional, never completion order, which is what makes it safe to zip the results back against your input list.

Partial Failure: the Argument That Actually Matters

Once three upstreams are involved, the probability that all three are healthy is lower than the probability that any one of them is. What gather does by default is therefore the first thing to get right.

By default, the first exception propagates to the caller the moment it occurs:

@app.get("/gather-fail-fast")
async def gather_fail_fast() -> dict:
    """Default gather: the first exception propagates and the good results are lost."""
    started = time.perf_counter()
    try:
        await asyncio.gather(call_upstream("prefs", 0.10), flaky(), call_upstream("orders", 0.20))
    except UpstreamError as exc:
        return {
            "elapsed_s": round(time.perf_counter() - started, 2),
            "raised": f"{type(exc).__name__}: {exc}",
            "results_kept": 0,
        }
    return {"unreachable": True}

$ GET /gather-fail-fast
200 OK
{
  "elapsed_s": 0.05,
  "raised": "UpstreamError: billing returned 503",
  "results_kept": 0
}

It returned after 0.05 s — the moment flaky() raised — and results_kept is 0. Worse than the lost data is what the transcript does not show: prefs and orders were not cancelled. They are still running, detached, and if one of them later raises you get an "exception was never retrieved" warning from a task nobody owns.

return_exceptions=True changes the contract completely. gather now waits for everything and hands you exceptions as values:

@app.get("/gather-partial")
async def gather_partial() -> dict:
    """return_exceptions=True: failures arrive as values, so the good data survives."""
    started = time.perf_counter()
    names = ["prefs", "billing", "orders"]
    raw = await asyncio.gather(
        call_upstream("prefs", 0.10),
        flaky(),
        call_upstream("orders", 0.20),
        return_exceptions=True,
    )
    ok, failed = {}, {}
    for name, item in zip(names, raw):
        if isinstance(item, BaseException):
            failed[name] = f"{type(item).__name__}: {item}"
        else:
            ok[name] = item
    return {
        "elapsed_s": round(time.perf_counter() - started, 2),
        "ok": ok,
        "failed": failed,
    }

$ GET /gather-partial
200 OK
{
  "elapsed_s": 0.2,
  "ok": {
    "prefs": {
      "upstream": "prefs",
      "took_s": "0.1"
    },
    "orders": {
      "upstream": "orders",
      "took_s": "0.2"
    }
  },
  "failed": {
    "billing": "UpstreamError: billing returned 503"
  }
}

Two real results plus one classified failure, in 0.2 s. The endpoint can now answer with degraded data and a machine-readable note about what is missing, which is almost always better for the caller than a 500.

Two details that bite people here. Results are BaseException instances, not raised — so you must test with isinstance and you will get no traceback unless you log one yourself. And return_exceptions=True swallows asyncio.CancelledError too, which means a gather in a cancelled request can quietly keep going; if your handler needs to be cancellable, prefer a task group.

asyncio.TaskGroup: Structured Concurrency

On Python 3.11+, asyncio.TaskGroup is the better default whenever a failed branch makes the whole operation pointless. It guarantees that no task outlives the block and cancels siblings on the first error:

@app.get("/taskgroup")
async def taskgroup() -> dict:
    """asyncio.TaskGroup (3.11+): a sibling failure cancels the rest and raises ExceptionGroup."""
    started = time.perf_counter()
    collected: dict[str, dict] = {}

    async def record(name: str, delay: float) -> None:
        collected[name] = await call_upstream(name, delay)

    errors: list[str] = []
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(record("prefs", 0.10))
            tg.create_task(record("profile", 0.30))
            tg.create_task(flaky())
    except* UpstreamError as eg:
        errors = [f"{type(e).__name__}: {e}" for e in eg.exceptions]
    return {
        "elapsed_s": round(time.perf_counter() - started, 2),
        "exception_group": errors,
        "completed_before_cancel": sorted(collected),
    }

$ GET /taskgroup
200 OK
{
  "elapsed_s": 0.05,
  "exception_group": [
    "UpstreamError: billing returned 503"
  ],
  "completed_before_cancel": []
}

completed_before_cancel is empty, and that is the point: prefs was 50 ms from finishing when flaky() raised at 0.05 s, and the task group cancelled it rather than letting it run on unobserved. Compare that with the fail-fast gather above, which returned at the same 0.05 s but left two tasks alive.

Note the syntax constraints, because they surprise people. except* matches by member type and always hands you an ExceptionGroup, so you iterate eg.exceptions. You cannot return, break or continue from inside an except* block — the code above assigns to errors and returns afterwards for exactly that reason. And a bare except ExceptionGroup also works if you do not need the filtering.

Timeouts: One Budget, or One per Call

A fan-out is only as reliable as its slowest member, so a deadline is mandatory. asyncio.timeout is a context manager, which means it wraps whatever shape of concurrency you already have.

Around the whole fan-out, it enforces a single endpoint-level budget:

@app.get("/timeout")
async def timeout() -> dict:
    """asyncio.timeout wraps the whole fan-out in one deadline."""
    started = time.perf_counter()
    try:
        async with asyncio.timeout(0.35):
            await asyncio.gather(call_upstream("prefs", 0.10), slow())
    except TimeoutError as exc:
        return {
            "elapsed_s": round(time.perf_counter() - started, 2),
            "budget_s": 0.35,
            "raised": type(exc).__name__,
            "detail": repr(str(exc)),
        }
    return {"unreachable": True}

$ GET /timeout
200 OK
{
  "elapsed_s": 0.35,
  "budget_s": 0.35,
  "raised": "TimeoutError",
  "detail": "''"
}

It returned at exactly the budget. Two things in that transcript matter operationally. The exception is a plain builtin TimeoutError (on 3.11+, asyncio.TimeoutError is an alias of it), and its message is the empty string — so any log line or error response you build must add the context itself, because the exception carries none.

The same primitive applied per call turns a hard failure into graceful degradation:

@app.get("/timeout-per-call")
async def timeout_per_call() -> dict:
    """A per-call deadline lets the fast upstreams still answer while the slow one is dropped."""
    started = time.perf_counter()

    async def with_budget(coro, name: str):
        try:
            async with asyncio.timeout(0.35):
                return await coro
        except TimeoutError:
            return {"upstream": name, "timed_out": True}

    results = await asyncio.gather(
        with_budget(call_upstream("prefs", 0.10), "prefs"),
        with_budget(slow(), "reports"),
    )
    return {"elapsed_s": round(time.perf_counter() - started, 2), "results": results}

$ GET /timeout-per-call
200 OK
{
  "elapsed_s": 0.35,
  "results": [
    {
      "upstream": "prefs",
      "took_s": "0.1"
    },
    {
      "upstream": "reports",
      "timed_out": true
    }
  ]
}

Same 0.35 s wall clock, but the caller gets real preferences data plus an explicit timed_out marker instead of nothing at all. Both patterns compose: an outer budget for the endpoint's contract, inner budgets so one bad dependency cannot consume it.

Bounding Concurrency with a Semaphore

The last failure mode is the one that only appears in production. If each request fans out to N upstream calls and you serve R requests per second, the downstream sees N x R concurrent connections. gather will happily open all of them.

@app.get("/semaphore/{limit}")
async def semaphore(limit: int) -> dict:
    """Bound concurrency: 8 calls of 100ms through a semaphore of `limit`."""
    sem = asyncio.Semaphore(limit)
    peak = 0
    inflight = 0

    async def guarded(i: int) -> int:
        nonlocal peak, inflight
        async with sem:
            inflight += 1
            peak = max(peak, inflight)
            await asyncio.sleep(0.10)
            inflight -= 1
            return i

    started = time.perf_counter()
    await asyncio.gather(*[guarded(i) for i in range(8)])
    return {
        "limit": limit,
        "calls": 8,
        "elapsed_s": round(time.perf_counter() - started, 2),
        "peak_in_flight": peak,
    }

$ GET /semaphore/8
200 OK
{
  "limit": 8,
  "calls": 8,
  "elapsed_s": 0.1,
  "peak_in_flight": 8
}

$ GET /semaphore/2
200 OK
{
  "limit": 2,
  "calls": 8,
  "elapsed_s": 0.4,
  "peak_in_flight": 2
}

With eight tokens, all eight ran together and the batch took 0.1 s. With two, peak_in_flight never exceeded 2 and the batch took 0.4 s — four sequential pairs. The semaphore did not slow anything down; it made the queue explicit and put it somewhere you control instead of in the downstream's accept backlog.

A caveat that costs people an afternoon: an asyncio.Semaphore created at module scope binds to whichever event loop first awaits it. Under a multi-worker deployment each process has its own loop and therefore its own semaphore, so the effective global limit is your per-worker limit times the worker count. Size accordingly, and create per-request semaphores (as above) or loop-scoped ones in the lifespan rather than sharing module state across loops.

Verification

Timing assertions are the only honest test of concurrency, and they belong in your suite:

import asyncio
import time

import httpx


async def test_fan_out_overlaps():
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://t") as c:
        body = (await c.get("/gather")).json()
    # Must equal the slowest branch, not the sum of the branches.
    assert body["elapsed_s"] <= body["slowest_upstream_s"] + 0.05


async def test_partial_failure_still_returns_data():
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://t") as c:
        body = (await c.get("/gather-partial")).json()
    assert set(body["ok"]) == {"prefs", "orders"}
    assert "billing" in body["failed"]

Write the assertion against a threshold that a serial implementation could not possibly meet, so a regression that removes the concurrency fails loudly. Structuring these around real HTTP fakes rather than mocks is covered in Testing FastAPI Applications.

Trade-offs and When Not To

A fan-out multiplies your blast radius. Three upstreams at 99.9% availability each give roughly 99.7% for the composite if all three are required. return_exceptions=True plus per-call timeouts is what buys that back, and the decision of which upstreams are required is a product decision, not a technical one.

Concurrency is not free at the connection layer. Share one httpx.AsyncClient across the fan-out so calls reuse pooled connections; creating a client per call means a fresh TLS handshake per call and often erases the gain entirely.

Do not fan out over unbounded input. asyncio.gather(*[fetch(u) for u in urls]) is fine for three URLs and a denial-of-service against yourself for thirty thousand. Bound it with a semaphore, or move the work out of the request entirely — see Background Task Processing.

If the data is stable, cache instead. The fastest upstream call is the one you did not make; Caching Strategies often beats any amount of concurrency tuning.

Do not gather blocking functions. gather only overlaps things that yield. A list of synchronous calls wrapped in coroutines still runs serially on the loop — they need Running Sync Code in a Threadpool first.

FAQ

What does return_exceptions=True actually change? Without it, gather re-raises the first exception immediately and you lose every result that had already arrived. With it, gather always waits for all awaitables and returns exceptions as ordinary values in the result list, so you can build a partial response from the calls that succeeded.

Should I use asyncio.gather or asyncio.TaskGroup? Use TaskGroup when a failure in any branch should abandon the whole operation, because it cancels the siblings for you and cannot leak an orphaned task. Use gather with return_exceptions=True when a partial result is genuinely useful to the caller.

Why does TaskGroup raise ExceptionGroup instead of my exception? Because several tasks can fail at once, TaskGroup collects every error into an ExceptionGroup rather than arbitrarily picking one. Handle it with except* clauses, which match by member type and give you the matching sub-group.

Does gather stop the other calls when one fails? No. gather propagates the first exception to the caller but the remaining tasks keep running in the background, which is how orphaned tasks and unretrieved exception warnings appear. TaskGroup cancels its siblings, which is why it is the safer default.

Where should the timeout go, around the whole fan-out or each call? Around the whole fan-out when the endpoint has one latency budget it must not exceed, and around each call when slow upstreams should be dropped individually so the fast ones still contribute. The two compose, and a per-call budget is what lets you degrade gracefully instead of failing entirely.

How do I stop a fan-out from overwhelming a downstream service? Wrap each call in an asyncio.Semaphore sized to the downstream's capacity so only that many run at once. The total work still completes, but in batches, which converts an unbounded connection spike into predictable queuing you control.