FastAPI – async def or def, and How to Tell

Whether a route is def or async def is the most consequential one-word decision in FastAPI, and the framework's name pushes people toward the wrong one. This lesson explains what actually happens in each case, and then measures it — including a case in this codebase where the concurrent version is two and a half times slower.

What FastAPI does with each

The framework accepts both and treats them completely differently.

@app.get("/sync")
def sync_route():
    return {"ok": True}       # dispatched to a WORKER THREAD


@app.get("/async")
async def async_route():
    return {"ok": True}       # runs directly ON THE EVENT LOOP

A def route is handed to a threadpool, so it can block for as long as it likes without affecting anything else. FastAPI does this deliberately, and it is why plain synchronous code is safe here in a way it is not in most async frameworks.

An async def route runs on the event loop, in the same thread as every other concurrent request. It gives control back only when it awaits. Between awaits, nothing else in the process runs.

That asymmetry produces the single most important rule:

If you are not going to await something, write def. An async def that never awaits is strictly worse than the same function written def — it takes the same time and blocks everything else while doing it.

The mistake, and why it is invisible

@app.get("/listings")
async def list_listings(db: Session = Depends(get_db)):
    # ⚠️ Synchronous database call inside an async route.
    return db.execute(select(Property).limit(20)).scalars().all()

SQLAlchemy's synchronous execute blocks. Inside async def it blocks the event loop, so while that query runs, no other request in the process makes progress — not the ones already mid-flight, not new ones arriving.

The reason it survives review is that it works perfectly in development. One developer, one request at a time, a 2 ms local query: indistinguishable from correct. Under concurrency it degrades in a way that looks like a database problem rather than a code problem, because every request slows down together.

The same trap covers anything synchronous: requests.get(), time.sleep(), open().read() on a large file, bcrypt, an image resize. All fine in def; all poison in async def.

Three ways out, in order of preference: write def and let the threadpool handle it; use a genuinely async library and await it; or push the blocking call to a thread explicitly:

from starlette.concurrency import run_in_threadpool


@app.get("/listings")
async def list_listings(db: Session = Depends(get_db)):
    rows = await run_in_threadpool(lambda: db.execute(select(Property).limit(20)).scalars().all())
    return rows

The third exists mostly for when you are already in an async route for another reason. If the whole function is one blocking call, the first option is better and simpler.

Concurrency is not parallelism

Worth separating, because the GIL confuses the two.

Concurrency is making progress on many things by interleaving them. One thread, one core, switching whenever something waits. That is what an event loop does, and it is what async gives you.

Parallelism is doing many things at literally the same instant, which needs more than one core. In CPython that needs more than one process, because the GIL means one thread executes Python bytecode at a time.

So neither async nor the threadpool makes CPU-bound work faster. Resizing images or computing a report will not speed up by being awaited — and in an async route it will freeze every other request while it runs. The answers there are a process pool, a worker service, or a library that releases the GIL in C.

from concurrent.futures import ProcessPoolExecutor

pool = ProcessPoolExecutor(max_workers=4)


@app.post("/reports")
async def build_report(spec: ReportSpec):
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(pool, expensive_pure_function, spec.model_dump())

Note it is handed a plain dict rather than a model — arguments cross a process boundary by pickling, and that is the same discipline background tasks needed for a different reason.

Also note what uvicorn's --workers actually gives you: separate processes, and therefore real parallelism across cores. Async buys concurrency within a worker; workers buy parallelism across them. Both, and they solve different problems.

The threadpool has a size

"Just use def" has a limit, and it is worth knowing before you meet it. The threadpool defaults to 40 threads. Forty concurrent def requests occupy all of them, and the forty-first waits — not for the database, for a thread.

That ceiling is usually fine, because it tends to sit above the database's own connection limit. It becomes the bottleneck when handlers are slow for reasons other than the database — a third-party HTTP call with a two-second latency, for instance. Twenty requests per second against a two-second call is exactly forty threads.

Raise it if you must, at startup:

import anyio


@asynccontextmanager
async def lifespan(_: FastAPI):
    anyio.to_thread.current_default_thread_limiter().total_tokens = 100
    yield

The better answer at that point is usually to make the slow call properly async, since waiting on a network is exactly what async is good at.

Now measure it

StayHub's /admin/stats runs eight independent aggregate queries. Nothing depends on anything else, so it is the textbook case for concurrency.

The synchronous version issues them one after another. The async version runs them together:

    (
        total_users, total_hosts, total_properties, published_properties,
        total_bookings, confirmed_bookings, cancelled_bookings, gross,
    ) = await asyncio.gather(

Textbook says eight serial round trips become one. Here is what actually happened, 30 runs each after warm-up:

/admin/stats        (8 serial)      median   7.9ms
/admin/stats-async  (8 concurrent)  median  19.8ms      <- 2.5x WORSE

The concurrent version is two and a half times slower, and both return identical numbers. That is not a bug in the implementation. It is what async costs when there is nothing to wait for.

Where the crossover is

Vary only the per-query wait — same eight queries, same machine, pg_sleep standing in for latency — and the picture resolves:

per-query wait   sync serial   async gather
          0ms         4.0ms        15.5ms     async 289% slower
          5ms        59.1ms        22.2ms     async  63% faster
         50ms       444.9ms        76.2ms     async  83% faster
        200ms      1654.7ms       232.9ms     async  86% faster

The rule falls out of the table: async removes waiting, not work. It charges a fixed overhead — a session and a pooled connection per task, the greenlet bridge SQLAlchemy's async layer uses, event-loop scheduling — and pays you back in proportion to how long you would otherwise have waited.

Break-even is roughly 1–2 ms of wait per call. Below it you are paying overhead for nothing. Above it the returns are large and approach the theoretical eight-fold ceiling.

Which side of the line things sit on:

StayHub's /admin/stats stays synchronous, because it is the faster endpoint. The async twin exists to be measured against, which is the only reason to keep two versions of anything.

The async database path

If you do want one, it is less work than expected:

async_engine = create_async_engine(
    settings.database_url,
    pool_pre_ping=True,
    echo=False,
    pool_size=5,
    max_overflow=5,
)

AsyncSessionLocal = async_sessionmaker(
    bind=async_engine, expire_on_commit=False, autoflush=False
)

Same URL and same driver — postgresql+psycopg:// is psycopg 3, which speaks both protocols. With psycopg 2 you would need a second driver, asyncpg, with its own type handling, which is how a Decimal starts arriving as a float on one code path and not the other.

Three things to watch.

The install needs an extra. SQLAlchemy's async layer is a greenlet bridge over the sync core, so it needs greenlet — which a bare install does not pull in:

SQLAlchemy[asyncio]==2.0.36

Without it everything imports fine and the first async query dies at close time with ValueError: the greenlet library is required to use this function, pointing at SQLAlchemy's internals rather than at anything you wrote.

The pools add up. An async engine alongside a sync one is a second pool, and Postgres counts total connections. Workers × (sync pool + async pool) has to stay under max_connections.

One session cannot multiplex. asyncio.gather over a single AsyncSession does not work — a session is one connection. Each task takes its own:

    async def scalar(stmt) -> int:
        # Its own session per task. See the ⚠️ above.
        async with AsyncSessionLocal() as s:
            return (await s.execute(stmt)).scalar_one() or 0

Which is also where a chunk of that fixed overhead comes from.

Where async is unambiguously right

Two cases in this application, neither about speed.

File uploads, because the API is async:

            while chunk := await file.read(CHUNK):

You cannot await in a def, so the route has to be async def. Reading file.file synchronously instead would block the loop for the duration of the upload — on a slow connection, seconds.

Many independent slow calls, which is the case the table above endorses. Three third-party APIs at 300 ms each is 900 ms serial and 300 ms concurrent, and that is a difference a user feels:

async with httpx.AsyncClient(timeout=5) as client:
    weather, events, transit = await asyncio.gather(
        client.get(f"{WEATHER}/forecast", params={"city": city}),
        client.get(f"{EVENTS}/upcoming", params={"city": city}),
        client.get(f"{TRANSIT}/lines", params={"city": city}),
    )

Note httpx.AsyncClient rather than requests. requests is synchronous; there is no way to await it, and calling it in an async route is the blocking mistake from the top of this lesson.

gather also has a sharp edge worth knowing: by default one failure cancels the others and propagates immediately. Pass return_exceptions=True when partial results are better than none — which for three independent widgets on a page they usually are.

The tools worth knowing

Four asyncio primitives cover almost everything a web application needs.

gather runs awaitables concurrently and returns their results in order. By default one failure cancels the rest and propagates; pass return_exceptions=True when partial results beat none:

results = await asyncio.gather(*calls, return_exceptions=True)
ok = [r for r in results if not isinstance(r, Exception)]

A timeout, because a request with no deadline is a worker held forever by somebody else's outage:

async with asyncio.timeout(2.0):
    data = await client.get(url)

A semaphore, to bound concurrency. Firing five hundred concurrent requests at a third-party API gets you rate-limited or blocked; ten at a time is usually as fast and does not:

sem = asyncio.Semaphore(10)


async def fetch(url: str):
    async with sem:
        return await client.get(url)

A task group, when several things must run together and any failure should cancel the others:

async with asyncio.TaskGroup() as tg:
    weather = tg.create_task(get_weather(city))
    events = tg.create_task(get_events(city))

One rule that is easy to break: do not create a bare task and forget it. asyncio.create_task(...) without holding a reference can be garbage-collected mid-flight, and its exception disappears with it. Keep the reference, or use a task group.

Streaming, which only async can do

Async is not only about speed. Some response shapes are unavailable without it, because they depend on sending in pieces:

from fastapi.responses import StreamingResponse


@app.get("/exports/bookings.csv")
async def export_bookings(db: AsyncDbSession):
    async def rows():
        yield "id,check_in,check_out,total\n"
        async for booking in stream_bookings(db):
            yield f"{booking.public_id},{booking.check_in},{booking.check_out},{booking.total}\n"

    return StreamingResponse(rows(), media_type="text/csv")

The alternative builds the whole export in memory before sending a byte, which for a large table is both a memory spike and a long silence the client may time out on. Streaming starts immediately and uses constant memory.

The same mechanism underlies server-sent events and WebSockets — both are long-lived connections, which is exactly what ASGI added over WSGI and exactly what a synchronous worker cannot hold.

The library you reach for changes

An async route needs async libraries. Reaching for the familiar synchronous one is the blocking mistake wearing a different hat:

Create HTTP clients once at startup rather than per request. An httpx.AsyncClient holds a connection pool, and building one per request throws away the pooling that makes it worth using:

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient(timeout=10)
    yield
    await app.state.http.aclose()

Mixing is fine

An application does not have to pick. StayHub is overwhelmingly synchronous with two async routes, and nothing special is required to combine them:

• An async dependency can serve a def route.
• A def dependency can serve an async def route — it runs in the threadpool.
• Middleware is always async, and can wrap either.
• Background tasks follow the same rule as routes: def for blocking work.

The one thing you cannot do is call an async function from a sync one without a loop. If you find yourself reaching for asyncio.run() inside a route, something has been declared the wrong way round.

Testing async code

TestClient handles async routes with no ceremony — it runs the loop for you, so a test looks identical whichever way the route is written. Testing an async function directly needs a plugin:

import pytest


@pytest.mark.asyncio
async def test_stats_async_matches_sync(db):
    a = stats(admin, db)
    b = await stats_async(admin, async_db)
    assert a == b

That assertion — that both versions return identical numbers — is the one worth writing whenever you keep two implementations. StayHub's benchmark script asserts it too, because a faster endpoint returning different answers is not a win:

    if a == b:
        print("  both endpoints return identical numbers ✓")
        return 0

Two async testing traps. Fixtures and tests must share an event loop, which is what asyncio_default_fixture_loop_scope in pytest.ini pins down — without it you get failures about a future attached to a different loop. And an async test that forgets to await passes silently, because an un-awaited coroutine is truthy: assert some_async_call() asserts that a coroutine object exists.

How to tell you got it wrong

Blocking the event loop has a distinctive signature, which is worth recognising because it does not look like a code problem.

Latency rises for everything at once. Not one endpoint — all of them, including the health check, including endpoints that touch nothing. That symmetry is the tell: a slow database makes database endpoints slow, while a blocked loop makes everything slow.

Throughput plateaus far below what one core should do. A blocked loop serialises every request in the worker, so the process handles one at a time regardless of concurrency.

It is fine in development. One request at a time never reveals it.

The direct way to confirm it is asyncio's own debug mode, which logs any callback that occupies the loop too long:

PYTHONASYNCIODEBUG=1 uvicorn app.main:app

Executing <Task ...list_listings...> took 0.412 seconds

That line names the coroutine. It is the fastest route from "the API is slow" to the specific function holding the loop, and it is one environment variable.

Two more things that surprise people

A dependency can be a different flavour from its route. An async dependency serving a def route is fine, and so is the reverse — a def dependency used by an async def route runs in the threadpool. So adopting async does not mean converting everything at once, which is what makes an incremental migration possible.

The GIL is not the reason async exists. Async is about not waiting; the GIL is about not executing bytecode in parallel. They are unrelated problems with unrelated solutions — async for I/O, processes for CPU — and conflating them leads to the belief that async will speed up a report generator. It will not.

Migrating a sync codebase, if you must

The wrong way is to add async to every route and fix the fallout. The order that works:

1. Find the endpoints that actually wait. Per-route latency percentiles, and then what they are waiting on. Most applications have one or two.
2. Convert one, with its libraries. An async route calling a sync client is worse than the sync route it replaced.
3. Measure it. If the wait is under a millisecond or two, revert — the table above says you have just made it slower.
4. Leave the rest alone. def routes in a threadpool are correct, not a compromise.

StayHub is the end state of exactly that process: two async routes, one because UploadFile requires it and one kept only for measurement, and everything else plain def. A mostly-synchronous FastAPI application is not a failure to adopt the framework properly. It is usually the right answer.

How to decide

1. Does the handler await anything? No → def. Done.
2. Does it await exactly one thing? Then async buys you concurrency with other requests, not within this one — worth it if the wait is long, since a thread is not tied up. Below a millisecond or two, not worth it.
3. Does it await several independent things? This is the case async is for. Use asyncio.gather.
4. Is anything in it blocking? Then it must be def, or the blocking part must go through run_in_threadpool.

And whichever you choose: measure before believing. The table above was produced by a thirty-line script against a real database, and it contradicted the expected answer. That is normal, and it is why the script lives in the repository next to the endpoints it compares.

Next: middleware, ordering and CORS — the layer wrapped around every route in this lesson, and an ordering rule that reads backwards.