Async Correctness and Concurrency in FastAPI

Async correctness is the discipline of keeping a FastAPI worker's single event loop free: choosing async def and def deliberately, never running synchronous work on the loop, and putting explicit bounds on the concurrency a free loop makes possible.

It is the foundation of Async, Background Tasks and Observability, and every other topic in that area depends on it. Async database sessions exist because a synchronous driver is the most common way to block the loop. Background task processing exists because some work should not be on the loop at all. Caching reduces how often you need to touch either.

The thing that makes this topic worth its own guide is that nothing in Python or FastAPI enforces any of it. async def is a promise the runtime never checks. Break it and you get no error, no warning, and a service that behaves perfectly until it has users.

Prerequisites

• FastAPI 0.139.2 on Python 3.12, served by an ASGI server such as Uvicorn.
• A working mental model of await as suspend here and let something else run, rather than as wait.
• Enough load-testing capability to issue genuinely concurrent requests. Every claim in this guide is invisible at concurrency 1.

Core Mechanics: One Loop, One Thread, No Preemption

A FastAPI worker process runs one event loop on one thread. Concurrency comes from coroutines voluntarily handing control back at await points, which lets the loop advance another coroutine while the first waits. That is the entire mechanism, and two consequences follow from it.

The loop cannot interrupt a running coroutine. There is no preemption, no time slice, no scheduler that steps in. A coroutine that does not suspend runs to completion, and while it does, the loop is inert. Every other in-flight request — for unrelated endpoints, from unrelated users — is frozen behind it.

The declaration decides the dispatch path, once. When you register a route, FastAPI inspects the endpoint and picks its strategy permanently: coroutine functions are awaited on the loop, plain functions are handed to a worker thread. That decision is made from the declaration alone, at import time. FastAPI cannot see what is inside the body, which is why declaring async def and then blocking is not caught by anything.

So the correct handler is one where the declaration and the body agree:

import httpx


@app.get("/upstream")
async def upstream() -> dict:
    """Correct: every wait in this body is an await, so the loop is free during it."""
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data", timeout=5.0)
    return response.json()

And an equally correct handler for work that has no async form:

@app.get("/legacy-report")
def legacy_report() -> dict:
    """Also correct: declared def, so Starlette runs the whole body on a worker thread."""
    return {"rows": sync_driver.query("SELECT ...")}

The full decision, with measured throughput for each combination and the microsecond cost of the threading hop, is worked through in FastAPI async def vs def.

Your Worker Has Three Ceilings, Not One

"How many concurrent requests can a worker handle?" has three different answers depending on which path the work takes, and knowing which one you are near is most of capacity planning. All three measured against the same 100 concurrent requests with a 50ms unit of work:

@app.get("/awaited")
async def awaited() -> dict:
    """Pure awaited I/O. The loop imposes no fixed ceiling on how many of these overlap."""
    await asyncio.sleep(WORK)
    return {"ok": True}


@app.get("/threaded")
def threaded() -> dict:
    """A def handler. Concurrency here is capped by the AnyIO capacity limiter."""
    time.sleep(WORK)
    return {"ok": True}


@app.get("/downstream")
async def downstream() -> dict:
    """An awaited call that a semaphore deliberately bounds, to protect something fragile."""
    async with DOWNSTREAM_LIMIT:          # asyncio.Semaphore(4)
        await asyncio.sleep(WORK)
    return {"ok": True}

Real output from running all three:

$ GET /probe/worker-budget
200 OK
{
  "unit_of_work_s": 0.05,
  "awaited_io": {
    "requests": 100,
    "waves": 1,
    "ceiling": "none imposed by the framework"
  },
  "def_handlers": {
    "requests": 100,
    "waves": 3,
    "ceiling": 40
  },
  "semaphore_bounded": {
    "requests": 100,
    "waves": 25,
    "ceiling": 4,
    "observed_peak_in_flight": 4
  }
}

Awaited I/O: one wave. A hundred overlapping awaits cost roughly what one costs. There is no framework-imposed limit here at all — the real ceiling is memory and whatever the other end of the connection can take.

def handlers: three waves. A hundred requests through a 40-token limiter is ceil(100 / 40) = 3. The limiter is a cap on concurrent offloaded calls rather than a pre-spawned pool; the mechanism and how to resize it are covered in Running Sync Code in a Threadpool.

Semaphore-bounded: twenty-five waves, with an observed peak of exactly 4 in flight. This one is deliberate, and the point is that it is predictable. You chose 4, you got 4, and the queue is explicit rather than emergent.

The fourth case is the one not in the transcript: a blocking call inside async def has an effective ceiling of one, and that one applies to the entire worker rather than to a single endpoint.

Production Implementation: Offloading and Bounding

Two patterns cover almost every real case.

Offload work that cannot be awaited. When an async def handler must call something synchronous — a legacy client, an image resize, a CPU-heavy hash — hand it to a thread so the loop stays free:

import anyio.to_thread


@app.post("/thumbnail")
async def thumbnail(image: bytes) -> dict[str, int]:
    # The loop is free while this runs on a worker thread.
    size = await anyio.to_thread.run_sync(resize_image, image)
    return {"bytes": size}

Threads are for blocking I/O. For CPU-bound work the GIL means threads buy you nothing, and the work belongs in a process pool or, better, on a queue.

Bound work that fans out. A free loop will happily open ten thousand connections to a service that can handle fifty. Concurrency that is possible is not concurrency that is wise:

import asyncio

# Sized to what the downstream can actually take, not to what we can generate.
_limit = asyncio.Semaphore(10)


async def fetch(client: httpx.AsyncClient, url: str) -> bytes:
    async with _limit:
        return (await client.get(url)).content

One caveat that costs people an afternoon: a module-scope asyncio.Semaphore binds to whichever event loop first awaits it, and it bounds one worker rather than your deployment. Four workers with a limit of 10 each present 40 concurrent calls to the downstream. The fan-out patterns, partial failure handling and timeout budgets are covered in Concurrent Requests with asyncio.gather.

Async and Performance Notes

Offloading moves a bottleneck; it does not remove one. The threadpool's 40 tokens are shared by every def handler, every def dependency and every explicit offload in the process. Saturate it and requests queue for a token, which looks exactly like a slow downstream in your metrics and is not.

Sync work that holds a database connection multiplies its cost. A threaded handler that checks out a connection ties up two scarce resources at once, and your effective thread count becomes bounded by your pool size. See Fixing asyncpg Connection Pool Exhaustion.

Prefer async libraries on the hot path. Offloading should be the exception for calls you cannot replace, not the standard way to reach I/O. Every offload costs a thread hop and consumes a token that something else might need.

Workers multiply everything. Every ceiling above is per worker. Four workers give you four loops, four 40-token limiters and four copies of every module-scope semaphore — which is good for throughput and bad for any downstream that assumed your limit was global.

Testing Strategy

The single most important thing to understand about testing async correctness is that a single request proves nothing. A blocking handler and a correct one take identical time to serve one request. Every test that issues one request at a time will pass on both.

So the tests that matter fan out:

async def test_endpoint_does_not_serialise(client):
    """Ten concurrent 100ms requests should overlap, not queue."""
    started = time.perf_counter()
    await asyncio.gather(*[client.get("/report") for _ in range(10)])
    elapsed = time.perf_counter() - started
    assert elapsed < 0.3, f"took {elapsed:.2f}s — something in the handler is blocking"

Three practical notes. Use httpx.AsyncClient with ASGITransport, not TestClient — TestClient calls block the calling thread and serialise, so a concurrency assertion through it is meaningless. Choose a threshold in the wide gap between "overlapping" and "serial" so the test does not flake on a loaded CI machine. And apply it to any endpoint whose I/O layer you do not fully control, because the case this catches is a dependency upgrade quietly swapping an async call for a sync one.

The broader client and fixture choices are covered in Testing FastAPI Applications, and the async-specific mechanics in Testing Async Endpoints with pytest-asyncio.

Failure Modes and Diagnosis

Unrelated endpoints slow down together. The signature of a blocked loop, because the loop is what they share. If only one endpoint is slow, look at that endpoint's dependencies instead.

Latency climbs linearly with concurrency. The definitive fingerprint. A correct endpoint's latency is flat as concurrency rises; a blocked one's is a straight line. Fixing Blocking Calls in Async Routes turns that curve into a line number.

Everything is fine until it is catastrophically not. Blocking degrades non-linearly. The worker copes until arrival rate exceeds serial capacity, then the queue grows without bound and latency runs away.

Hidden blocking inside a library. An SDK with async def methods may wrap a synchronous client. Async-looking is not async. Measure a single call in isolation with a heartbeat next to it.

Threadpool saturation. Symptoms look like a slow downstream: requests waiting, nothing obviously busy. Graph borrowed_tokens against total_tokens; if you are at the ceiling, the limiter is setting your p99.

Unbounded fan-out. asyncio.gather over a large list opens that many connections at once. Fine for three URLs, an attack on yourself for thirty thousand.

A sync database driver. Still the most common single cause, and the reason Async Database Sessions is a topic of its own.

Choosing an Approach

FAQ

Should my route be async def or plain def? Use async def when every wait in the body is awaited, and plain def when any line blocks, because FastAPI dispatches def handlers to a worker thread where blocking is harmless. The one combination to avoid is an async def handler containing a blocking call, which stalls every other request on that worker.

Why does one blocking call slow down unrelated requests? Each worker runs a single event loop on one thread. While that thread is inside a synchronous call it cannot advance any other coroutine, so every concurrent request on that worker waits. The loop only achieves concurrency when coroutines yield at await points, which blocking calls never do.

How many concurrent requests can one worker actually handle? It depends which path they take. Awaited I/O has no framework-imposed ceiling: 100 concurrent requests completed in one wave. Plain def handlers are capped by a 40-token limiter and took three waves. Semaphore-bounded work takes as many waves as you configured.

How do I run blocking code without blocking the loop? Offload it with anyio.to_thread.run_sync or Starlette's run_in_threadpool for blocking I/O, and to a process pool for CPU-bound work. The coroutine awaits the offloaded result, so the loop stays free while the work runs elsewhere.

How do I limit concurrency so I do not overwhelm a downstream service? Guard the calls with an asyncio.Semaphore sized to what the downstream can take. Measured with a limit of 4, one hundred requests ran as twenty-five waves with an observed peak of exactly 4 in flight, converting an unbounded burst into a predictable queue.

How do I detect blocking before it reaches production? Write concurrency tests rather than single-request tests. A blocking handler and a correct one serve one request in identical time, so only overlapping requests can tell them apart. In production, a heartbeat coroutine measuring event-loop lag catches what tests cannot.

Related

• Up to the area: Async, Background Tasks and Observability.
• The decision: FastAPI async def vs def Performance measures all four combinations of declaration and workload.
• The diagnosis: Fixing Blocking Calls in Async Routes goes from a latency curve to the offending line.
• The offload mechanism: Running Sync Code in a Threadpool covers the limiter, the helpers and how to resize the pool.
• Using the free loop: Concurrent Requests with asyncio.gather on fan-out, partial failure and timeout budgets.
• Where it usually goes wrong: Async Database Sessions, the most common source of loop-blocking calls.