Search Topics

Search across all FastAPI topics

GitHub

Async Endpoints

Core Concept

You added async to your endpoint and made it slower. Let's figure out why.

You write async def read_items() and call requests.get() inside it. Your API handles one request at a time. 100 concurrent users? They wait in a single-file queue. You used async but got worse performance than sync.

terminal
# Your "async" endpoint:
@app.get("/items")
async def read_items():
    response = requests.get("https://api.example.com/data")  # ← BLOCKING!
    return response.json()

# Load test results:
# Sync endpoint (def):     100 req in 2.1s ✓
# Your async endpoint:     100 req in 45.3s ✗
# Why is async SLOWER?!

Question

Wait, isn't async supposed to be faster? Why did adding one keyword make your API 20x slower? The answer comes down to one thing: what happens inside your function when it hits that blocking call.

How FastAPI Handles Your Functions

FastAPI treats async def and def very differently under the hood. Here's what happens when a request comes in:

With async def:

Request arrives

FastAPI receives it

Runs on event loop

Single thread, shared

Hits await

Pauses, lets others run

I/O completes

Resumes where it left off

With plain def:

Request arrives

FastAPI receives it

Sent to thread pool

Gets its own thread

Blocks thread

Only this thread waits

Returns result

Thread freed up

async vs def routing

What you just learned

async def runs directly on the event loop — it must use await to yield control

Plain def runs in a separate thread pool — blocking is safe because it only blocks that thread

The disaster: async def + blocking call = blocks the entire event loop

The Trap: async def + Blocking Code

Here's the code that ruined your load test. It looks async, but it's lying. The requests library is synchronous. It doesn't know what await is. It just... blocks.

main.py
import requests  # ← This is the problem

@app.get("/items")
async def read_items():
    # This blocks the ENTIRE event loop!
    # No other request can be processed while waiting
    response = requests.get("https://api.example.com/data")
    return response.json()

# Meanwhile, 99 other users are waiting...
# The event loop is frozen on this one HTTP call.

The Fix: Use Async Libraries

If you're going to use async def, every I/O call inside it needs to be awaitable. Replace requests with httpx.

main.py
import httpx  # ← async-ready HTTP client

@app.get("/items")
async def read_items():
    async with httpx.AsyncClient() as client:
        # This yields to the event loop while waiting
        response = await client.get("https://api.example.com/data")
        return response.json()

# Now 100 users can all be "waiting" at the same time.
# The event loop juggles them all.

blocking vs non-blocking

What you just learned

requests.get() inside async def blocks the entire event loop — all users freeze

httpx.AsyncClient with await lets the event loop handle other requests while waiting

If you can't use async libraries, just use plain def — it's safer

The Real Power: Concurrent I/O

Here's where async really shines. Need data from three APIs? Don't wait for each one sequentially — fire them all at once with asyncio.gather.

main.py
import asyncio
import httpx

@app.get("/dashboard")
async def get_dashboard():
    async with httpx.AsyncClient() as client:
        # Fire all three requests at the same time
        users, orders, stats = await asyncio.gather(
            client.get("https://api.example.com/users"),
            client.get("https://api.example.com/orders"),
            client.get("https://api.example.com/stats"),
        )
    # Total time ≈ slowest request, not sum of all three!
    return {
        "users": users.json(),
        "orders": orders.json(),
        "stats": stats.json(),
    }

Go Deeper: Running Blocking Code Safely

Sometimes you're stuck with a blocking library. Maybe it's a legacy SDK, or CPU-intensive image processing. You can offload it to a thread pool from an async endpoint using run_in_threadpool.

main.py
from starlette.concurrency import run_in_threadpool

def cpu_intensive_task(data: list) -> dict:
    # Heavy computation — no async version available
    result = process(data)
    return result

@app.post("/process")
async def process_data(data: list[int]):
    # Offload to thread pool — event loop stays free
    result = await run_in_threadpool(cpu_intensive_task, data)
    return result

Insight

This is exactly what FastAPI does automatically for plain def endpoints. So if your entire endpoint is blocking code, just use def instead of async def — less boilerplate, same result.

advanced patterns

What you just learned

asyncio.gather runs multiple async operations concurrently — total time equals the slowest, not the sum

run_in_threadpool offloads blocking code to a thread from an async context

If everything in your endpoint is blocking, just use def — FastAPI handles threading for you

See It: Sync vs Async Race

Watch 3 identical database queries run side by side. Sync processes them one at a time. Async overlaps the I/O waits, finishing all three in the time it takes sync to do one.

Think about it...

If you define def my_endpoint() (no async), does FastAPI block the event loop when it runs?

Hint: Think about what FastAPI does differently for def vs async def.

Key Points

async def

For I/O-bound work with async libraries (httpx, asyncpg)

plain def

For CPU-bound work or blocking libraries — runs in thread pool

asyncio.gather

Run multiple async operations concurrently for speed

run_in_threadpool

Offload blocking code to a thread from an async context