Async Endpoints with async/await

Your FastAPI endpoint is running, but every time it calls a database or an external API, the whole server freezes. Other requests pile up, response times balloon, and your carefully built service suddenly feels like it's running on a 1990s dial-up connection. This is the classic blocking-I/O trap, and it's exactly the pain point that building async endpoints with async/await solves. By the end of this lesson, you'll be able to write endpoints that handle thousands of concurrent connections without breaking a sweat — turning your API from a single-lane road into a multi-lane highway.

The problem this lesson solves

When you write a standard def endpoint in FastAPI, the server executes it in a thread pool. That's fine for quick, CPU-bound calculations. But the moment your endpoint does something that waits — a database query, an HTTP call to a third-party service, reading a large file — that thread sits idle, blocked, doing absolutely nothing while it waits for a response.

With the default thread pool, each blocked thread is one fewer slot available for other requests. Under load, your server runs out of threads, and new requests start queuing up. The result: latency spikes, dropped requests, and a poor user experience. The deeper issue is that I/O-bound operations waste the most valuable resource your server has — time. A single slow database call can bring your entire API to its knees.

Pro Tip: The enemy isn't slowness — it's blocking. An endpoint that takes 1 second to return is fine if it can handle 100 of those requests simultaneously. The problem is when that 1-second wait blocks 99 other requests from even starting.

Core concept / mental model

Think of your FastAPI server as a restaurant kitchen. In the traditional (synchronous) model, you have a fixed number of chefs (threads). When one chef is waiting for the oven to preheat, they can't do anything else — they're stuck, staring at the timer. If all chefs are waiting on ovens, the kitchen is effectively closed.

Async/await changes the game. Now imagine your chef is event-driven. When they put something in the oven, they write a note: "Check this in 2 minutes." Then they immediately move on to the next task. When the timer rings, they come back and finish the dish. One chef can now manage dozens of dishes simultaneously because they never wait — they delegate.

In Python terms:

• async def marks a function as a coroutine — a function that can be paused and resumed.
• await tells Python: "I'm about to do something slow. Free me up to do other work until this comes back."
• FastAPI runs a single event loop that juggles all these paused coroutines, resuming each one when its awaited I/O completes.

The beauty is that you write code that looks synchronous — no callbacks, no threading headaches — but behaves concurrently.

How it works step by step

When a request hits an async FastAPI endpoint, here's what happens under the hood:

1. Request arrival: The ASGI server (like Uvicorn) receives the HTTP request and hands it to the event loop.
2. Coroutine creation: FastAPI calls your async def endpoint, which returns a coroutine object (it hasn't run yet).
3. Execution begins: The event loop starts running your coroutine.
4. Hit await: Your coroutine executes until it hits an await statement (e.g., await db.fetch()). At this point, it suspends — essentially saying "I'll wait for this, but don't hold up anyone else."
5. Context switch: The event loop immediately picks up another pending request and starts processing it.
6. I/O completes: When the database (or external API, or file system) responds, the event loop receives a notification.
7. Resumption: The event loop resumes your coroutine right where it left off, with the result of the awaited operation.
8. Response: Your endpoint completes, FastAPI serializes the response, and the server sends it back to the client.

The key insight: at no point is the server thread blocked. While your endpoint awaits the database, the event loop is doing useful work for other requests.

Pro Tip: await only works on awaitable objects — usually other coroutines (things defined with async def) or objects that implement __await__. You can't await a regular synchronous function.

Hands-on walkthrough

Let's build a practical example. We'll create a FastAPI app with a slow, synchronous endpoint and a fast, asynchronous one, then compare how they behave under load.

Project setup

First, make sure you have FastAPI and an ASGI server installed:

pip install fastapi uvicorn httpx aiohttp

Your first async endpoint

Create a file async_app.py:

import asyncio
from fastapi import FastAPI

app = FastAPI(title="Async Demo")

async def fetch_data_from_db(user_id: int):
    """Simulates a slow database query."""
    await asyncio.sleep(2)  # Pretend this is a real DB call
    return {"user_id": user_id, "data": "expensive query result"}

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    """Async endpoint — can handle many concurrent requests."""
    data = await fetch_data_from_db(user_id)
    return {"status": "success", **data}

Run it with:

uvicorn async_app:app --reload

Now, while that server is running, you can make multiple requests concurrently:

import asyncio
import httpx

async def make_request(client, idx):
    resp = await client.get(f"http://localhost:8000/users/{idx}")
    print(f"Request {idx}: status={resp.status_code}, body={resp.json()}")

async def main():
    async with httpx.AsyncClient() as client:
        # Fire off 5 requests at the same time
        await asyncio.gather(*(make_request(client, i) for i in range(5)))

asyncio.run(main())

You'll see all 5 requests complete in roughly 2 seconds total (not 10 seconds, as a synchronous server would take). The event loop interleaves all five 2-second sleeps simultaneously.

Mixing sync and async

FastAPI lets you mix both styles. This is often what you'll do in a real app:

import asyncio
from fastapi import FastAPI

app = FastAPI(title="Mixed Sync/Async")

# CANNOT use await here — this is a regular function

def compute_heavy_task(data: list[int]) -> int:
    """Pure CPU work — runs in a thread pool."""
    return sum(x * x for x in data)

# CAN use await here — this is an async endpoint

@app.get("/report")
async def get_report():
    # Quick computation (doesn't need async)
    numbers = list(range(1000))

    # CPU-bound work — FastAPI runs this in a thread pool automatically
    result = await run_in_threadpool(compute_heavy_task, numbers)

    # I/O-bound work — truly async
    external_data = await fetch_external_api()

    return {"result": result, "external": external_data}

Important distinction:

• async def endpoints — run on the event loop. Good for I/O-heavy work. Must not contain blocking calls.
• def endpoints — run in a thread pool. Good for CPU-bound work. Perfect when your endpoint uses blocking libraries (like requests or a synchronous ORM).

Async database calls

Real async shines with async database drivers. Here's a typical pattern with asyncpg and SQLAlchemy async:

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy import text

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"
engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession)

app = FastAPI()

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

@app.get("/products/{product_id}")
async def get_product(product_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(text("SELECT * FROM products WHERE id = :id"), {"id": product_id})
    product = result.fetchone()
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")
    return {"product": dict(product)}

The await db.execute(...) suspends the coroutine while the database processes the query — freeing the event loop for other requests. To run this, you'd need SQLAlchemy's async support installed (pip install sqlalchemy[asyncio] asyncpg).

Pro Tip: Look for async-native libraries. For HTTP, use httpx or aiohttp instead of requests. For databases, find the async driver (e.g., asyncpg for PostgreSQL, aiomysql for MySQL).

Compare options / when to choose what

Not every endpoint should be async def. Knowing when to use what is a core skill.

Variations to consider:

• asyncio.to_thread() — Python 3.9+, lets you run a blocking function in a separate thread from within a coroutine (FastAPI's run_in_threadpool() does this internally).
• HTTPX vs aiohttp — HTTPX has a cleaner API and supports HTTP/2; aiohttp is battle-tested for high-throughput scenarios.
• Starlette's run_in_threadpool — when you need to call a blocking library from an async endpoint and still keep the event loop responsive (e.g., requests or SyncMySql driver).

Troubleshooting & edge cases

"TypeError: 'coroutine' object is not subscriptable"

You forgot await. If you call an async function but don't await it, you get a coroutine object, not the result.

# Wrong — returns a coroutine object, not data

data = fetch_data_from_db(42)

# Right

data = await fetch_data_from_db(42)

Blocking the event loop with synchronous code

Sometimes people mix sync libraries into async endpoints:

@app.get("/bad")
async def bad_endpoint():
    resp = requests.get("https://api.example.com")  # BLOCKS the whole event loop
    return resp.json()

This blocks everything. Every other request waits until this returns. Fix: Use httpx.AsyncClient or wrap in run_in_threadpool:

from starlette.concurrency import run_in_threadpool

@app.get("/good")
async def good_endpoint():
    resp = await run_in_threadpool(requests.get, "https://api.example.com")
    return resp.json()

"Event loop is already running"

If you try asyncio.run() inside an endpoint that's already on the event loop, you'll get this error. Instead, use await directly:

# Wrong

asyncio.run(some_coroutine())

# Right

await some_coroutine()

Memory leaks with opening async clients per request

Creating a new httpx.AsyncClient() for every request creates many connections and leaks sockets. Fix: Create one client at startup and reuse it via FastAPI's dependency injection:

from fastapi import FastAPI, Depends
import httpx

app = FastAPI()

async def get_client():
    async with httpx.AsyncClient() as client:
        yield client

@app.get("/weather")
async def get_weather(client: httpx.AsyncClient = Depends(get_client)):
    resp = await client.get("https://api.weather.example/current")
    return resp.json()

"Task was destroyed but it is pending"

If you create background tasks with asyncio.create_task() and don't keep a reference, they may get garbage-collected. Keep references or use FastAPI's BackgroundTasks.

What you learned & what's next

You've now mastered building async endpoints with async/await in FastAPI. Let's recap the core takeaways:

• async def endpoints run on a single event loop and can handle thousands of concurrent I/O-bound requests without blocking.
• await suspends a coroutine without blocking the loop — the server does other work while waiting.
• I/O-bound operations (DB calls, HTTP requests, file reads) belong in async endpoints with async-native libraries.
• CPU-bound operations belong in regular def endpoints (FastAPI uses a thread pool) or delegated via run_in_threadpool.
• Mixing async drivers (asyncpg, httpx, aiofiles) is essential — blocking libraries destroy async benefits.

The next lesson in this track covers Background Tasks and Lifespan Events. You'll learn how to run work after a response is sent (like sending emails or processing uploads) and how to manage startup/shutdown resources like database connection pools. Your async knowledge will be crucial there, as background tasks are built directly on FastAPI's async foundation.