Identifying the Blocking Behavior

A common pitfall in FastAPI occurs when a synchronous, blocking operation is placed directly inside a coroutine endpoint. This causes the event loop thread to freeze, preventing other requests from being processed until the blocking call finishes. Consider the following minimal reproduction, where a simulated delay runs inside an async function:

import time
import uvicorn
from fastapi import FastAPI

api = FastAPI()


def heavy_job(seconds):
    print("heavy_job started")
    time.sleep(seconds)
    print("heavy_job completed")


@api.get("/alpha")
async def endpoint_alpha():
    print("alpha invoked")
    return "alpha response"


@api.get("/beta")
async def endpoint_beta():
    start = int(time.time())
    print(f"beta start: {start}")
    heavy_job(8)
    end = int(time.time())
    elapsed = end - start
    print(f"beta end: {end}")
    return f"beta took: {elapsed}s"

Simultaneous requests to /alpha and /beta reveal that /alpha only completes after the 8‑second heavy_job finishes. The log order and measured response times confirm this serialization:

• /beta starts first.
• heavy_job blocks the thread for 8 seconds.
• /alpha is processed only afterward, itself taking around 8 seconds.

Solution 1: Remove async from the Blocking Endpoint

The most straightforward fix is to declare the endpoint as a plain synchronous function. FastAPI runs synchronous path operasions in a separate thread pool, which keeps the main event loop free:

@api.get("/beta")
def endpoint_beta_sync():
    start = int(time.time())
    print(f"beta start: {start}")
    heavy_job(8)
    end = int(time.time())
    print(f"beta end: {end}")
    return f"beta took: {end - start}s"

With this change, the /alpha endpoint receives an immediate response regardless of the heavy work happening concurrently.

Solution 2: Offload Work via run_in_executor

When an async endpoint cannot be converted to synchronous (e.g., it performs other non‑blocking I/O in parallel), you can push the CPU‑bound or blocking call into a thread or process executor using asyncio.get_event_loop().run_in_executor():

import asyncio
import time

@api.get("/beta")
async def endpoint_beta_async():
    start = int(time.time())
    print(f"beta start: {start}")
    loop = asyncio.get_running_loop()
    await loop.run_in_executor(None, heavy_job, 8)
    end = int(time.time())
    print(f"beta end: {end}")
    return f"beta took: {end - start}s"

The executor delegates the blocking heavy_job to a separate thread while the async function suspends at the await. The event loop remains available to serve other inbound requests concurrently.

Key Takeaway

Embedding synchronous, long‑running calls directly inside an async def endpoint without await serializes request processing. Always either move such endpoints to plain synchronous functions or offload blocking tasks with await and an appropriate executor to maintain FastAPI’s high concurrency.