cubic
Try cubic for free

Prevent Blocking Calls in Async Python Code

Finds time.sleep, requests, sync database drivers, and heavy CPU work inside async def functions, where one slow call stalls every task on the event loop.

Catch this in your code

What it flags

Flaggedapp/routes/reports.py
import time

import requests
from fastapi import FastAPI, HTTPException

app = FastAPI()


@app.get("/reports/{report_id}")
async def get_report(report_id: str) -> dict:
    for _ in range(10):
        # Both calls block the event loop and stall every other request on this worker
        resp = requests.get(f"https://reports.internal/jobs/{report_id}", timeout=5)
        resp.raise_for_status()
        job = resp.json()
        if job["status"] == "done":
            return job
        time.sleep(1)
    raise HTTPException(status_code=504, detail="Report is still running")
Passesapp/routes/reports.py
import asyncio

import httpx
from fastapi import FastAPI, HTTPException

app = FastAPI()
client = httpx.AsyncClient(base_url="https://reports.internal", timeout=5)


@app.get("/reports/{report_id}")
async def get_report(report_id: str) -> dict:
    for _ in range(10):
        resp = await client.get(f"/jobs/{report_id}")
        resp.raise_for_status()
        job = resp.json()
        if job["status"] == "done":
            return job
        await asyncio.sleep(1)
    raise HTTPException(status_code=504, detail="Report is still running")

Why it matters

An event loop runs one callback at a time, so a 200 ms blocking call inside a coroutine delays every other task on that loop by up to 200 ms. The code works in development with one user, then latency spikes and timeouts appear under concurrent load. FastAPI makes this easy to miss: a plain def endpoint runs in a threadpool, while the same code in an async def endpoint runs on the loop itself.

Agent instructions

cubic checks every pull request against these instructions. Paste them into AGENTS.md or Cursor rules and your AI coding tool follows them while it writes code.

AGENTS.md

Add this section to AGENTS.md at the repository root. Codex, Cursor, and other AI coding tools read it.

## Prevent Blocking Calls in Async Python Code

Applies to `**/*.py`.

Review `async def` functions, including asyncio tasks and async web handlers such as FastAPI or Starlette endpoints, for calls that block the event loop.

Flag changes that:
- Call `time.sleep()` inside a coroutine instead of `await asyncio.sleep()`.
- Make HTTP calls with `requests`, `urllib.request`, or a sync `httpx.Client` inside a coroutine instead of `httpx.AsyncClient`, `aiohttp`, or another async client.
- Use a sync database or cache driver inside a coroutine, such as `psycopg2`, `sqlite3`, `pymysql`, a sync SQLAlchemy `Session`, or `redis.Redis`, instead of an async driver or `asyncio.to_thread()`.
- Run CPU-heavy work inline, such as password hashing, image processing, large JSON or CSV parsing, or compression; offload it with `asyncio.to_thread()` when the library releases the GIL, or with a `ProcessPoolExecutor` through `loop.run_in_executor()`.
- Call `subprocess.run()`, `os.system()`, or blocking reads and writes of large files inside a coroutine; use `asyncio.create_subprocess_exec()` or offload the work.
- Block on thread primitives inside a coroutine, such as `threading.Lock.acquire()`, `queue.Queue.get()`, or `concurrent.futures.Future.result()`.
- Change a FastAPI endpoint or dependency from `def` to `async def` while it still calls blocking libraries.

Do not flag:
- Plain `def` FastAPI endpoints and dependencies, or sync Starlette endpoints, which run in a threadpool.
- Blocking calls wrapped in `asyncio.to_thread()`, `loop.run_in_executor()`, `anyio.to_thread.run_sync()`, or `run_in_threadpool()`.
- Small, fast operations such as reading a small config file at startup, logging, or parsing a small JSON payload.

Checks files matching **/*.py.

More Python agents

See all Python code review standards

Run this agent on every pull request

cubic reviews every pull request against the agents you turn on and comments on the lines that break them.

Catch this in your code