KAHIBARO
Discord Login Register ☰
☰

Async Endpoints

Table of Contents

Why Async Endpoints Matter in FastAPI

When you build backends that talk to databases, call other APIs, or do file operations, your code often waits for I/O. If your server waits in a blocking way, it cannot handle many requests at the same time.

FastAPI is built on top of asyncio and is designed to work very well with asynchronous endpoints. In this chapter you will learn how to write async endpoints correctly, when to use async def versus def, what you can and cannot do in async functions, and how this affects performance.

The goal here is not to teach all of async programming in Python, but to show how to use it correctly inside FastAPI.

Sync vs Async Path Operations

In FastAPI, you can define path operations using both normal functions and async functions.

python 
from fastapi import FastAPI
app = FastAPI()
# Synchronous endpoint
@app.get("/sync")
def get_sync():
    return {"type": "sync"}
# Asynchronous endpoint
@app.get("/async")
async def get_async():
    return {"type": "async"}

FastAPI can call both kinds of functions. The difference is how they behave when there is waiting for I/O.

What happens inside each type

Type of functionDefinitionCan use awaitBehavior with I/O
SynchronousdefNoBlocking. The worker waits during I/O.
Asynchronousasync defYesNon-blocking if you await async I/O operations.

In an async endpoint:

python 
@app.get("/hello")
async def hello():
    # this is allowed
    await some_async_call()
    return {"msg": "hello"}

In a sync endpoint:

python 
@app.get("/hello-sync")
def hello_sync():
    # this is NOT allowed here
    # await some_async_call()   # SyntaxError
    return {"msg": "hello"}

FastAPI decides how to execute your function based on its signature:

A First Async Endpoint

Here is a minimal full example with one async and one sync endpoint that simulate work with time.sleep and asyncio.sleep.

python 
from fastapi import FastAPI
import time
import asyncio
app = FastAPI()
@app.get("/slow-sync")
def slow_sync():
    time.sleep(3)  # blocking sleep
    return {"status": "done", "type": "sync"}
@app.get("/slow-async")
async def slow_async():
    await asyncio.sleep(3)  # non-blocking sleep
    return {"status": "done", "type": "async"}

If you send many requests at the same time:

You will see the performance difference clearly if you use a load testing tool.

Using async and await in FastAPI

Inside an async endpoint, you can and should use await for any operation that supports it.

Example with asyncio.sleep and an async HTTP client:

python 
import asyncio
import httpx
from fastapi import FastAPI
app = FastAPI()
@app.get("/external")
async def call_external():
    await asyncio.sleep(0.5)  # simulate some delay
    async with httpx.AsyncClient() as client:
        response = await client.get("https://httpbin.org/get")
    data = response.json()
    return {"external_origin": data["origin"]}

Key points:

Example of a mistake:

python 
@app.get("/bad")
async def bad():
    # This creates a coroutine object but never runs it
    asyncio.sleep(1)  # missing 'await'
    return {"ok": True}

This endpoint will return immediately. The sleep never actually happens.

When to Use async def vs def

You can always use def, even in FastAPI. Using async def only helps if you use non-blocking I/O inside the function.

Use this rule:

Rule:
Use async def for endpoints that call async libraries (for example async DB drivers, async HTTP clients, async file I/O).
Use def for pure CPU work or when using only blocking libraries.

Good use cases for async endpoints

Example calling an async database method:

python 
from fastapi import FastAPI
app = FastAPI()
# imagine this is your async repository:
class UserRepo:
    async def get_user(self, user_id: int):
        # await some db call here
        ...
user_repo = UserRepo()
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = await user_repo.get_user(user_id)
    return user

When async does not help

If your function:

Then using async def does not make it faster. In fact, it can make things worse if you use blocking calls inside async def.

Example of what not to do:

python 
import time
from fastapi import FastAPI
app = FastAPI()
@app.get("/wrong")
async def wrong():
    # This blocks the event loop for 5 seconds
    time.sleep(5)
    return {"msg": "this hurts concurrency"}

In this example, the worker cannot handle other requests during the time.sleep.

If you must run blocking code but you still want an async endpoint, move the blocking work into a separate thread or process. You will see that in the next section.

Blocking Code in Async Endpoints

Sometimes you have no choice but to use a blocking library. For example, a blocking database driver or a machine learning library that takes CPU time.

You have three main options:

  1. Keep the endpoint synchronous with def.
  2. Use a background task system (see dedicated chapter).
  3. Run the blocking code in a thread from the async endpoint.

Option 1: Keep the endpoint sync

If your stack is mostly sync, you can simply write:

python 
@app.get("/report")
def generate_report():
    data = generate_report_sync()  # heavy sync function
    return data

FastAPI will run generate_report in a threadpool automatically, so it does not block the event loop. You lose some of the benefits of async, but the code is simpler.

Option 3: Run blocking code in a thread from async

If you want to keep the endpoint async, you can offload the blocking function to a thread.

From Python 3.9 onward, a common pattern is:

python 
import anyio
from fastapi import FastAPI
app = FastAPI()
def blocking_db_call():
    # blocking work, example with time.sleep
    import time
    time.sleep(2)
    return {"result": "done"}
@app.get("/threaded")
async def threaded():
    result = await anyio.to_thread.run_sync(blocking_db_call)
    return result

Older style with asyncio.get_running_loop().run_in_executor:

python 
import asyncio
from fastapi import FastAPI
app = FastAPI()
def blocking_db_call():
    import time
    time.sleep(2)
    return {"result": "done"}
@app.get("/threaded-old")
async def threaded_old():
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(None, blocking_db_call)
    return result

Now the event loop is free while your blocking code runs in another thread.

Using Async Database Libraries

A common real use case is an async database driver. For example, with SQLAlchemy 2.x async engine.

Very simplified example:

python 
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from .database import get_async_session
from .models import User
app = FastAPI()
@app.get("/users/{user_id}")
async def get_user(
    user_id: int,
    session: AsyncSession = Depends(get_async_session),
):
    result = await session.execute(
        select(User).where(User.id == user_id)
    )
    user = result.scalar_one_or_none()
    if not user:
        return {"error": "User not found"}
    return {"id": user.id, "email": user.email}

Points to notice:

This pattern allows FastAPI to handle many requests efficiently, even when each request talks to the database.

Async HTTP Clients in Endpoints

Calling external APIs is another typical task. With an async endpoint you can use an async HTTP client to avoid blocking.

Example using httpx.AsyncClient:

python 
import httpx
from fastapi import FastAPI
app = FastAPI()
@app.get("/weather/{city}")
async def get_weather(city: str):
    url = "https://api.example.com/weather"
    params = {"city": city}
    async with httpx.AsyncClient() as client:
        response = await client.get(url, params=params)
    data = response.json()
    return {
        "city": city,
        "temperature": data["temperature"],
        "description": data["description"],
    }

If your request handler needs to call multiple services, you can even call them concurrently using asyncio.gather.

Running Multiple Async Operations Concurrently

A big advantage of async is that you can start multiple I/O operations and wait for all of them at once.

Example: call two microservices in parallel from a single endpoint.

python 
import asyncio
import httpx
from fastapi import FastAPI
app = FastAPI()
async def get_user_profile(client: httpx.AsyncClient, user_id: int):
    r = await client.get(f"https://api.example.com/users/{user_id}")
    return r.json()
async def get_user_orders(client: httpx.AsyncClient, user_id: int):
    r = await client.get(f"https://api.example.com/users/{user_id}/orders")
    return r.json()
@app.get("/user-summary/{user_id}")
async def user_summary(user_id: int):
    async with httpx.AsyncClient() as client:
        profile_task = get_user_profile(client, user_id)
        orders_task = get_user_orders(client, user_id)
        profile, orders = await asyncio.gather(profile_task, orders_task)
    return {
        "user": profile,
        "orders": orders,
    }

Here:

Limitations and Gotchas with Async Endpoints

Async endpoints are powerful but there are some important rules and common mistakes.

Rule 1: Do not mix blocking calls in async code

Important:
Avoid calling blocking libraries such as time.sleep, requests.get, blocking DB drivers, or long CPU loops directly inside async def endpoints.
If you must, offload them to a thread.

Bad:

python 
import requests
@app.get("/bad-external")
async def bad_external():
    # This blocks the event loop
    response = requests.get("https://httpbin.org/get")
    return response.json()

Better:

python 
import requests
import anyio
def blocking_request():
    return requests.get("https://httpbin.org/get").json()
@app.get("/better-external")
async def better_external():
    data = await anyio.to_thread.run_sync(blocking_request)
    return data

Or best: use an async HTTP client instead of requests.

Rule 2: Never forget to await async calls

Bad:

python 
async def do_something():
    async_operation()  # missing await
    return "done"

Correct:

python 
async def do_something():
    await async_operation()
    return "done"

If you see a function defined with async def, it almost always should be awaited somewhere.

Rule 3: Do not call async from sync without a runner

Inside a plain def endpoint, you cannot call an async function directly.

Bad:

python 
def sync_endpoint():
    result = some_async_function()  # returns coroutine, does not run it
    return result

If you must call async from sync, you need something to run it, such as asyncio.run (but you usually should not call it in FastAPI endpoints) or a proper runner. In practice, you should design your code so that:

Patterns for Structuring Async Code

To keep your code clean:

python 
  class UserService:
      def __init__(self, session_factory):
          self.session_factory = session_factory
      async def get_user(self, user_id: int):
          async with self.session_factory() as session:
              user = await session.get(User, user_id)
              return user

Example:

python 
from fastapi import Depends
def get_user_service():
    return UserService(session_factory=make_session_factory())
@app.get("/users/{user_id}")
async def read_user(
    user_id: int,
    service: UserService = Depends(get_user_service),
):
    user = await service.get_user(user_id)
    if not user:
        return {"error": "not found"}
    return user

This keeps your endpoint thin and uses async in a clean way.

Simple Performance Experiment

You can build a tiny app to feel the difference between sync blocking and async non-blocking.

python 
# file: app.py
from fastapi import FastAPI
import time
import asyncio
app = FastAPI()
@app.get("/sync-sleep")
def sync_sleep():
    time.sleep(2)
    return {"status": "ok"}
@app.get("/async-sleep")
async def async_sleep():
    await asyncio.sleep(2)
    return {"status": "ok"}

Run the server with Uvicorn:

bash 
uvicorn app:app --workers 1

Then in another terminal, run something like:

bash 
# 10 requests to /sync-sleep
ab -n 10 -c 10 http://127.0.0.1:8000/sync-sleep
# 10 requests to /async-sleep
ab -n 10 -c 10 http://127.0.0.1:8000/async-sleep

Or use any other tool that can send concurrent requests.

You will see that with only 1 worker:

Summary

Async endpoints in FastAPI:

If you keep these rules in mind, you can build FastAPI applications that are both easy to read and highly performant under load.

Views: 8

Comments

Please login to add a comment.

Don't have an account? Register now!

© 2025 Kahibaro
Visitors: 3105 / 24h
Impressum / Imprint
Datenschutz / Data Policy