02 - Async/Await in Python: Understanding Event Loops, Coroutines, and I/O-Bound Concurrency

Putting async def in front of a FastAPI route does not make it faster: it makes it able to wait without occupying a thread. If you then call a blocking function inside that route (a time.sleep, a synchronous HTTP client, a traditional database driver), you stop the whole server until it finishes. I measured the difference on one hundred requests that each simulate fifty milliseconds of waiting, in nine consecutive runs: 0.09-0.16 s with async def and await asyncio.sleep, 0.18-0.23 s with a plain def, 5.09-5.20 s with async def and time.sleep. The ranges are wide because other processes were busy on the machine (load average between 13 and 16 on 12 cores); the proportions are not. The rest of the article explains why they are what they are.

The listings were run on 2026-11-12 with CPython 3.12.13, FastAPI 0.136.3 and httpx 0.28.1, on a 12-core machine; the line that starts with # Output: is the real output. Timings change from one machine and one run to the next, so for anything that depends on the machine I report an order of magnitude or a range, not an «expected» value. No server was started: requests to FastAPI go through httpx's ASGI transport, in the same process and without a network. It is not a production benchmark, it is an experiment that isolates the variable we care about.

The GIL is not the problem you think it is

The Global Interpreter Lock lets only one thread at a time execute Python bytecode (that is the definition in the official documentation's glossary). People often conclude that Python «cannot do several things at once». That is false for work that consists of waiting. A thread that sleeps or waits for a database response is not executing bytecode, and another thread can move on. What the GIL prevents is two threads doing computation at the same instant, and we will measure that in the last section.

An API spends most of its time waiting: the database, another service, a file. Imagine a waiter who takes an order at one table, carries it to the kitchen and, instead of staying there until the dish is ready, goes to the next table; he returns to the kitchen when called. That is what asyncio does: a single thread, a single waiter, and the rule that every long wait must be declared with await to allow the switch of table. Python 3.13 introduced an experimental build without the GIL (the release notes of version 3.13 say so), but everything that follows applies to the normal interpreter.

Event loop and coroutines, with a time to read

The event loop is the cycle that keeps the list of ready coroutines, runs one until it meets an await on a real wait, sets it aside and moves to the next. A coroutine is a function declared with async def: calling it does not run it, it produces an object that the loop can advance. The point that escapes beginners is that await on its own does not create concurrency: it suspends the current coroutine until the wait is over. Concurrency arises when several coroutines are waiting at the same time.

The listing compares two ways of reading three sources with latencies of 0.2, 0.3 and 0.1 seconds (the typical case of a page that must consult users, orders and preferences). In the sequential version each await waits for the previous one. In the asyncio.gather version the three coroutines start together and the total time is that of the slowest.

# sequence.py
import asyncio
import time


async def read(name: str, delay: float) -> str:
    await asyncio.sleep(delay)          # here the coroutine hands control back to the loop
    return f"{name} ({delay} s)"


async def sequential():
    users = await read("users", 0.2)
    orders = await read("orders", 0.3)
    preferences = await read("preferences", 0.1)
    return [users, orders, preferences]


async def concurrent():
    return await asyncio.gather(
        read("users", 0.2),
        read("orders", 0.3),
        read("preferences", 0.1),
    )


async def measure(function):
    start = time.perf_counter()
    result = await function()
    print(f"{function.__name__:14s} {time.perf_counter() - start:.2f} s  {result}")


async def main():
    await measure(sequential)
    await measure(concurrent)


asyncio.run(main())

# Output:
# sequential     0.60 s  ['users (0.2 s)', 'orders (0.3 s)', 'preferences (0.1 s)']
# concurrent     0.30 s  ['users (0.2 s)', 'orders (0.3 s)', 'preferences (0.1 s)']

The result is 0.60 against 0.30 seconds: the sum of the waits against the maximum. The timings of this experiment are almost exact because the waits are artificial; with a real database they will have a variability of their own. Note that the code of the read function is identical in the two cases: what changes is only the way you call it. If the three reads depended on one another (the orders need the user's id), the sequence would be mandatory and gather would not help.

The most common trap: the coroutine that doesn't start

Forgetting the await does not produce an error: it produces an object. The line result = read("users") has run nothing, and if you don't look at the returned type nobody warns you until the object is discarded. Python reports it with a RuntimeWarning, which ends up on stderr among the other messages and often goes unnoticed. The listing captures the warning to show it to you.

# trap.py
import asyncio
import gc
import warnings


async def read(name: str) -> str:
    await asyncio.sleep(0.1)
    return name


with warnings.catch_warnings(record=True) as caught:
    warnings.simplefilter("always")
    result = read("users")           # the await is missing: the function was NOT run
    print(type(result).__name__)
    del result
    gc.collect()

print(caught[0].category.__name__, "-", caught[0].message)
print(asyncio.run(read("users")))    # this works: the loop runs it

# Output:
# coroutine
# RuntimeWarning - coroutine 'read' was never awaited
# users

The type is coroutine, and the message is «coroutine 'read' was never awaited». It is the signature of a missed await. The last line shows the correct call: the coroutine must be handed to a loop, with await inside another coroutine or with asyncio.run at the top level. In FastAPI the typical case is forgetting await on a call to an async client inside the route: the response does not contain the data you expected, because you are holding the coroutine and not its result.

gather, create_task and TaskGroup: what happens when one fails

There are three ways to run several coroutines together, and the difference shows mostly when one of them fails:

• asyncio.gather(...): waits for all of them and returns the list of results in the order in which you passed them.
• asyncio.create_task(...): starts a coroutine in the background right away and gives you back a Task object to wait for later.
• asyncio.TaskGroup (Python 3.11 and later): an async with block that waits for all the tasks created inside it and stops them if one fails.

The criterion for choosing is error handling. With gather you want partial results, with TaskGroup you want all or nothing, with create_task you have a job that must outlive the function that starts it, and then you must keep a reference to the task: the documentation warns that the loop keeps only weak references, so a task without references can be discarded by the garbage collector before it finishes.

# errors.py
import asyncio


async def job(name: str, delay: float, fails: bool = False) -> str:
    await asyncio.sleep(delay)
    if fails:
        raise ValueError(f"{name} failed")
    print(f"  {name} completed")
    return name


async def with_gather():
    print("gather:")
    try:
        await asyncio.gather(job("A", 0.1, fails=True), job("B", 0.3))
    except ValueError as e:
        print("  raised:", e)
    await asyncio.sleep(0.4)             # time for B to finish, if it was not stopped


async def with_return_exceptions():
    print("gather with return_exceptions=True:")
    results = await asyncio.gather(
        job("A", 0.1, fails=True), job("B", 0.3), return_exceptions=True
    )
    print("  ", results)


async def with_taskgroup():
    print("TaskGroup:")
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(job("A", 0.1, fails=True))
            group.create_task(job("B", 0.3))
    except* ValueError as error_group:
        print("  raised:", [str(e) for e in error_group.exceptions])
    await asyncio.sleep(0.4)


async def main():
    await with_gather()
    await with_return_exceptions()
    await with_taskgroup()


asyncio.run(main())

# Output:
# gather:
#   raised: A failed
#   B completed
# gather with return_exceptions=True:
#   B completed
#    [ValueError('A failed'), 'B']
# TaskGroup:
#   raised: ['A failed']

Three behaviors, three results. With gather without options, A's exception arrives immediately to whoever is waiting, but B is not stopped: it continues and completes, as the documentation says («Other awaitables … won't be cancelled»). If you did not expect B to finish, you have a job running with nobody collecting its result. With return_exceptions=True no exception is raised: the list contains the error in place of A's result, and it is up to you to check it with isinstance. With TaskGroup, instead, B does not complete: the group cancels it as soon as A fails, and the exceptions come out grouped in an ExceptionGroup, which is caught with the except* syntax. It is the behavior to prefer when the group's jobs make sense only together; for the details of cancellation there is advanced async/await: TaskGroup and cancellation.

async def or def in a FastAPI route

FastAPI accepts both, and they do different things. An async def route is run directly in the event loop; a def route is run in a thread of a pool (the Concurrency and async / await page says so), so it does not block the loop. So neither of the two «is wrong»: what is wrong is putting the wrong wait inside. FastAPI's documentation gives three indications, and I report them in order:

• If the library you use has a version with await (httpx for HTTP, async drivers for databases), declare the route with async def and use await.
• If the library does not support await, like most traditional database clients, declare the route with a plain def.
• If you don't know what to do, use def; if the route waits for nothing, async def is fine even without await.

The thread that ties the three rules together is «don't put a blocking call inside the event loop». A def that calls a synchronous library is correct, because the block happens in a thread and the loop stays free; an async def that calls the same library is the mistake, because the block happens in the loop and makes all the other requests wait. The experiment that follows puts the three combinations side by side.

# bench.py
import asyncio
import time

import httpx
from anyio import to_thread
from fastapi import FastAPI

app = FastAPI()
DELAY = 0.05          # seconds of simulated "database wait"


@app.get("/sync")
def sync_blocking():                  # def: FastAPI runs it in a pool thread
    time.sleep(DELAY)
    return {"ok": True}


@app.get("/async")
async def async_non_blocking():       # async def: the wait hands control back to the loop
    await asyncio.sleep(DELAY)
    return {"ok": True}


@app.get("/async-wrong")
async def async_blocking():           # async def + blocking call: the case to avoid
    time.sleep(DELAY)
    return {"ok": True}


async def measure(label: str, path: str, n: int = 100):
    transport = httpx.ASGITransport(app=app)       # no network: the app runs in-process
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        start = time.perf_counter()
        responses = await asyncio.gather(*(client.get(path) for _ in range(n)))
        elapsed = time.perf_counter() - start
    ok = sum(r.status_code == 200 for r in responses)
    print(f"{label:24s} {n} requests in {elapsed:5.2f} s  (ok: {ok})")


async def main():
    limiter = to_thread.current_default_thread_limiter()
    print("default pool threads:", limiter.total_tokens)
    await measure("/async", "/async")
    await measure("/sync", "/sync")
    limiter.total_tokens = 100
    await measure("/sync with pool of 100", "/sync")
    limiter.total_tokens = 40
    await measure("/async-wrong", "/async-wrong")


asyncio.run(main())

# Output:
# default pool threads: 40
# /async                   100 requests in  0.11 s  (ok: 100)
# /sync                    100 requests in  0.22 s  (ok: 100)
# /sync with pool of 100   100 requests in  0.16 s  (ok: 100)
# /async-wrong             100 requests in  5.15 s  (ok: 100)

The three lines are the thesis of the article. With /async the hundred 50 ms waits all overlap: from 0.09 to 0.16 s across the nine runs (in the output above, one of them). With /sync FastAPI uses the thread pool, which by default has 40 slots (AnyIO's limit, also reported in the first line of the output, and described in the AnyIO documentation): a hundred requests in three waves, 3 × 0.05 s, that is 0.15 s ideal; measured from 0.18 to 0.23 s, with thread startup and machine load making up the rest. Raising the limit to a hundred (fourth line) the time comes closer to the async one. With /async-wrong every request stops the loop for 50 ms, none overlaps with the others, and the total is the sum: 100 × 0.05 = 5 seconds, measured from 5.09 to 5.20. This is the only one of the three values that does not depend on the machine's load, because it is dominated by the sum of the waits.

Two caveats on the experiment. The first: the difference between /async and /sync here is small because forty sleeping threads cost little; with many thousands of simultaneous connections the pool of 40 becomes the ceiling and the difference grows, but I did not measure it and I give no numbers. The second: time.sleep is the textbook case, but the same block is produced by requests.get or a query with psycopg2 inside an async def route. If the symptom is «the server answers one request at a time even with only one process free», look there.

CPU work: neither await nor threads

Async does not help when the time is spent computing. A computation loop has no points at which to hand over control: if it sits in the loop it occupies it until the end, and if it sits in a thread it collides with the GIL. The only way to use several cores with normal CPython is to use several processes. The listing measures three ways of running four identical computations of about 0.15-0.2 s each: directly in the loop, in threads with asyncio.to_thread, in processes with ProcessPoolExecutor. In parallel a «sentinel» coroutine runs that should wake up every 10 ms: its longest pause measures how long the loop stayed busy.

# cpu.py
import asyncio
import time
from concurrent.futures import ProcessPoolExecutor

N = 3_000_000
JOBS = 4          # four computations to run "together"


def compute(n: int) -> int:
    return sum(i * i for i in range(n))     # pure CPU work, no waiting


async def computations_in_loop():
    return [compute(N) for _ in range(JOBS)]      # direct calls: the loop stays busy


async def heartbeat(gaps: list[float], stop: asyncio.Event):
    last = time.perf_counter()
    while not stop.is_set():
        await asyncio.sleep(0.01)           # should come back every ~10 ms
        now = time.perf_counter()
        gaps.append(now - last)
        last = now


async def trial(name: str, run):
    gaps: list[float] = []
    stop = asyncio.Event()
    sentinel = asyncio.create_task(heartbeat(gaps, stop))
    await asyncio.sleep(0.05)
    start = time.perf_counter()
    await run()
    elapsed = time.perf_counter() - start
    stop.set()
    await sentinel
    print(f"{name:22s} took {elapsed:.2f} s, longest loop pause {max(gaps) * 1000:4.0f} ms")


async def main():
    loop = asyncio.get_running_loop()
    await trial("directly in the loop", computations_in_loop)
    await trial(
        "asyncio.to_thread",
        lambda: asyncio.gather(*(asyncio.to_thread(compute, N) for _ in range(JOBS))),
    )
    with ProcessPoolExecutor(max_workers=JOBS) as pool:
        await trial(
            "ProcessPoolExecutor",
            lambda: asyncio.gather(*(loop.run_in_executor(pool, compute, N) for _ in range(JOBS))),
        )


if __name__ == "__main__":
    asyncio.run(main())

# Output:
# directly in the loop   took 0.89 s, longest loop pause  903 ms
# asyncio.to_thread      took 1.18 s, longest loop pause  111 ms
# ProcessPoolExecutor    took 0.60 s, longest loop pause   26 ms

The listing's output is one of the nine runs. Read the two columns separately. The duration: in the loop and in threads it is practically the same (the GIL does not let the four threads compute together), with processes it drops sharply because the computations run on different cores. In nine consecutive runs I measured, for the three ways, durations from 0.59 to 0.92 s, from 0.73 to 1.19 s and from 0.22 to 0.60 s. The loop pause: in the loop it is almost equal to the duration (the server answers nobody for all that time), in threads it drops to a value between about 103 and 219 ms, with processes it stays between 16 and 26 ms. The thread therefore serves to keep the loop alive, not to go faster.

Processes have a cost: the function and its arguments must be serializable with pickle and the function must be importable by the child processes (the concurrent.futures documentation says so), hence the if __name__ == "__main__" check in the listing. For jobs of a few milliseconds the cost of moving the data exceeds the gain. In a FastAPI route the pool is created once at application startup, not on every request, and truly heavy work (model inference, video processing) is better put in a job queue, with the route returning an identifier right away.

Can I use requests inside an async def route?

No: requests is synchronous, and inside a coroutine it stops the loop for the whole duration of the call, as in the third route of the experiment. You have two correct alternatives: declare the route with def and let FastAPI run it in a thread, or replace the library with httpx.AsyncClient and use await. If you cannot replace it and the route must stay async, wrap the call in await asyncio.to_thread(...).

Is async def always faster than def?

No, and the experiment shows it: with overlapping waits the two approaches were in the same order of magnitude (0.09 s against 0.18 s), and the difference makes itself felt only when the pool of 40 threads saturates. The advantage of async def is the ability to keep many waits open with few resources; the price is the discipline required: a single missing await or a single blocking call ruins everything. That is why the documentation suggests def when you are not sure.

What you can do now

You can explain why the GIL does not prevent concurrency on network waits, tell a sequence of awaits from a gather and read the difference in seconds, and recognize a forgotten await from the message «was never awaited». You can choose among gather, create_task and TaskGroup based on what must happen when a job fails. And you can decide between def and async def in FastAPI with a rule that you check with a stopwatch: which library do I call, and does it support await.

The task that remains, before putting the service in production, is to describe and validate the data coming in and going out: Pydantic does that, and it is the subject of Pydantic v2: advanced validation. For the starting point, that is the structure of the service and the generated documentation, you will find FastAPI from scratch; to put a service with models in production, FastAPI and Uvicorn in production.