Python asyncio in Production: Pitfalls Every API Developer Hits

How the asyncio event loop runs your API on one thread

asyncio is Python's standard library framework for running many concurrent operations on a single thread. An async def function produces a coroutine. Coroutines run on an event loop, which keeps a queue of ready work and a registry of pending I/O, such as sockets waiting for data or timers waiting to fire.

Scheduling is cooperative. A coroutine runs without interruption until it reaches an await on something that isn't ready, such as a network read, a database reply or asyncio.sleep. At that point it yields control back to the loop. The loop then resumes some other coroutine whose I/O has completed. The loop has no way to interrupt a coroutine. It regains control only when the running code yields.

That gives you the core rule: nothing may block the loop. A blocking call never yields, so while it runs, every other in-flight request on that worker is frozen. Each of the following stalls the whole process:

• time.sleep(1) stops everything for one second.
• A requests call to a slow upstream holds the thread until the response arrives.
• A synchronous database driver blocks on each query.
• Parsing a large JSON payload or running a pathological regex is pure CPU work with no await point.
• Reading a file with plain open(). Ordinary file I/O is blocking, and the loop gets no notification when it finishes.

The model pays off for I/O-bound API services that spend most of each request waiting on networks, databases or other services. A single thread can hold thousands of concurrent waits cheaply. It does not help CPU-bound work, because there is nothing to wait on and so nothing to overlap.

Copyimport asyncio
import time

async def heartbeat(stop: asyncio.Event) -> None:
    # Stands in for other in-flight requests on this worker.
    # It should print roughly every 100 ms if the loop is free.
    start = time.perf_counter()
    while not stop.is_set():
        print(f'  tick at {time.perf_counter() - start:.2f}s')
        await asyncio.sleep(0.1)

async def blocking_handler() -> None:
    # Never yields: the whole loop, and every other coroutine, freezes.
    time.sleep(0.5)

async def async_handler() -> None:
    # Yields to the loop while waiting, so other work keeps running.
    await asyncio.sleep(0.5)

async def run(handler) -> None:
    stop = asyncio.Event()
    hb = asyncio.create_task(heartbeat(stop))
    await asyncio.sleep(0)  # let the heartbeat start first
    await handler()
    stop.set()
    await hb

async def main() -> None:
    print('blocking handler (time.sleep):')
    await run(blocking_handler)   # expect a ~0.5 s gap between ticks
    print('async handler (asyncio.sleep):')
    await run(async_handler)      # expect ticks every ~0.1 s

asyncio.run(main())

A heartbeat task shows that time.sleep inside a coroutine freezes all other work on the loop for its full duration, while asyncio.sleep lets other coroutines keep running.

Finding blocking calls before your users do

A blocking call rarely fails loudly. A synchronous HTTP client, a time.sleep, a CPU-heavy JSON parse or a driver without async support simply holds the thread. While it runs, every other coroutine on that loop waits, including requests that have nothing to do with the slow code.

Turn on asyncio debug mode

asyncio has a debug mode built for this. Enable it with the PYTHONASYNCIODEBUG=1 environment variable, with loop.set_debug(True), or by passing debug=True to asyncio.run. In debug mode the loop logs a warning through the asyncio logger whenever a single callback or task step runs longer than loop.slow_callback_duration. The default threshold is 100 ms. Lower it to match your latency budget. The warning identifies the offending callback, so you get a code location instead of a vague report that the service felt slow. Debug mode also flags coroutines that were never awaited, which is another common source of silent bugs.

Watch p99, not the mean

A stall delays every request that is in flight at that moment, but only for the length of the stall. If most requests finish quickly, the average stays low and hides the problem. The 99th percentile captures it, because the unlucky requests queued behind the blocked loop land there. A rising p99 with a flat median is a strong sign that something is holding the loop.

Use it deliberately

Debug mode adds overhead to every callback, so don't leave it on across a production fleet. Run it in staging under realistic load, or enable it on a small sample of instances. That gives you named slow callbacks to work from during incident triage, which is faster than combing through flame graphs or reading code line by line.

Copyimport asyncio
import logging
import time

# Debug-mode warnings are emitted through the "asyncio" logger.
logging.basicConfig(level=logging.WARNING)


async def handler_with_blocking_call():
    # Simulates a sync client or CPU-heavy parse: holds the thread
    # and stalls every other coroutine on this loop.
    time.sleep(0.2)


async def healthy_handler():
    # Yields to the loop, so it never trips the slow-callback check.
    await asyncio.sleep(0.2)


async def main():
    loop = asyncio.get_running_loop()
    # Default threshold is 0.1 s; tighten it to match your latency budget.
    loop.slow_callback_duration = 0.05
    await asyncio.gather(healthy_handler(), handler_with_blocking_call())


if __name__ == "__main__":
    # Equivalent to PYTHONASYNCIODEBUG=1 or loop.set_debug(True).
    # Use in staging or on sampled instances: it adds per-callback overhead.
    asyncio.run(main(), debug=True)
    # Expected log line, naming the offending task step:
    # WARNING:asyncio:Executing <Task ... coro=<handler_with_blocking_call() ...>>
    #   took 0.2xx seconds

Running with asyncio debug mode and a lowered slow_callback_duration makes the loop log a warning that names the coroutine whose blocking time.sleep held the event loop.

Offloading blocking work: threads, processes and their limits

Once you have a named blocking call, the quickest fix is to move it off the event loop. asyncio.to_thread() runs a function in the loop's default ThreadPoolExecutor and copies the current context, so context variables such as request IDs still reach your logging. loop.run_in_executor() does the same job but accepts an explicit executor. Use it to give a slow dependency its own pool so it cannot starve everything else.

The default pool is bounded, at min(32, CPU count + 4) workers on recent Python versions. Once every worker is busy, new calls wait in the executor's internal queue. That queue has no timeout and exposes no metrics, so latency grows while the event loop itself looks healthy. Cancelling the awaiting coroutine does not stop a thread that is already running, either.

For CPU-bound work, a ProcessPoolExecutor avoids the GIL, but the function and its arguments must be pickled for every call.

Offloading hides a blocking call; it does not remove it. If one sync-only database driver, SDK or ORM path sits on a hot route, every request on that route needs a thread. Your concurrency is then capped by pool size, not by the event loop, and you pay for async complexity while getting threaded throughput.

Copyimport asyncio
import hashlib
import time
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor

# Dedicated pool for a slow sync-only SDK, so its backlog cannot exhaust the default pool.
REPORTS_POOL = ThreadPoolExecutor(max_workers=8, thread_name_prefix="reports-sdk")

def fetch_profile(user_id: int) -> dict:
    time.sleep(0.2)  # stands in for a blocking database driver call
    return {"id": user_id}

def render_report(report_id: str) -> str:
    time.sleep(1.0)  # stands in for a slow vendor SDK call
    return f"report-{report_id}"

def hash_payload(data: bytes) -> str:
    # CPU-bound work. It must be a top-level function so the process pool can pickle it.
    for _ in range(200_000):
        data = hashlib.sha256(data).digest()
    return data.hex()

async def handle_request(cpu_pool: ProcessPoolExecutor) -> None:
    loop = asyncio.get_running_loop()
    # Default pool. to_thread copies contextvars, so request IDs reach the worker thread.
    profile = await asyncio.to_thread(fetch_profile, 42)
    # Explicit pool. When it is saturated, calls queue here, not in the default pool.
    # A timeout stops the await, but the thread keeps running until render_report returns.
    report = await asyncio.wait_for(loop.run_in_executor(REPORTS_POOL, render_report, "q3"), timeout=5)
    # Separate process avoids the GIL. Arguments and the result are pickled on every call.
    digest = await loop.run_in_executor(cpu_pool, hash_payload, b"payload")
    print(profile, report, digest[:16])

async def main() -> None:
    with ProcessPoolExecutor(max_workers=2) as cpu_pool:
        await asyncio.gather(*(handle_request(cpu_pool) for _ in range(4)))
    REPORTS_POOL.shutdown(wait=True)

if __name__ == "__main__":  # required for process pools under the spawn start method
    asyncio.run(main())

Three ways to offload work: asyncio.to_thread for blocking I/O on the default pool, run_in_executor with a dedicated ThreadPoolExecutor to isolate a slow sync-only SDK, and a ProcessPoolExecutor for CPU-bound work, with comments on where queuing, cancellation and pickling costs appear.

Task lifecycles: owning every task with TaskGroup

Fire-and-forget tasks can disappear

asyncio.create_task() schedules a coroutine and returns a Task, but the event loop keeps only a weak reference to it. If your code drops the return value, as in a handler that kicks off an audit write and returns, the task can be garbage-collected before it finishes. You get no error, only work that never completes. The documented fix is to keep a strong reference: add the task to a set and remove it in a done callback. Even then, nothing awaits that task, so its exceptions surface only as a "Task exception was never retrieved" log line, often long after the request is gone.

Structured concurrency with TaskGroup

asyncio.TaskGroup, added in Python 3.11, gives every task an owner. Tasks created through the group are tied to an async with block, and the block does not exit until all of them finish. If one task raises, the group cancels its siblings, waits for them to unwind, and raises the failures together as an ExceptionGroup, which you handle with except*. No task outlives the request that started it, and no failure goes unnoticed.

The trade-off is that failures become loud. With gather(), the first exception propagates but the other awaitables keep running. With return_exceptions=True, you get partial results and decide what counts as success. Teams that relied on that behavior, for example rendering a page when one of five backend calls fails, will see whole requests fail after migrating. If partial success is a real requirement, catch exceptions inside each task and return a sentinel value, so the policy is explicit.

TaskGroup semantics and related APIs have changed across releases. Check the asyncio documentation for the Python version you actually deploy (docs.python.org), not the latest one.

Copyimport asyncio

background_tasks: set[asyncio.Task] = set()

async def write_audit_log(event: str) -> None:
    await asyncio.sleep(0.1)  # stand-in for a database or queue write

def fire_and_forget(event: str) -> None:
    # The loop holds only a weak reference; keep a strong one until the task is done.
    task = asyncio.create_task(write_audit_log(event))
    background_tasks.add(task)
    task.add_done_callback(background_tasks.discard)

async def fetch(name: str, fail: bool = False) -> str:
    await asyncio.sleep(0.1)  # stand-in for a backend call
    if fail:
        raise ConnectionError(f"{name} unavailable")
    return f"{name}: ok"

async def fetch_optional(name: str, fail: bool = False) -> str | None:
    # Explicit partial-success policy: this call may fail without failing the request.
    try:
        return await fetch(name, fail)
    except ConnectionError:
        return None

async def handle_request() -> dict:
    fire_and_forget("profile_viewed")
    result: dict = {}
    try:
        async with asyncio.TaskGroup() as tg:  # Python 3.11+
            user = tg.create_task(fetch("users"))
            recs = tg.create_task(fetch_optional("recommendations", fail=True))
        # Reached only after every task in the group has finished.
        result = {"user": user.result(), "recs": recs.result()}
    except* ConnectionError as eg:
        # A required call failed: siblings were cancelled. No 'return' allowed in except*.
        result = {"errors": [str(e) for e in eg.exceptions]}
    return result

async def main() -> None:
    print(await handle_request())  # {'user': 'users: ok', 'recs': None}
    await asyncio.sleep(0.2)  # demo only: let the background audit write finish

asyncio.run(main())

Keeping a strong reference to a fire-and-forget task, using TaskGroup so a failing required call cancels its siblings and surfaces as an ExceptionGroup, and making partial success explicit by catching errors inside an optional task.

Cancellation, timeouts and bounded concurrency

Never swallow CancelledError

Cancellation is delivered as asyncio.CancelledError at the task's current await. It inherits from BaseException, so a bare except Exception won't catch it. A bare except: or except BaseException will, so re-raise it there. Swallowing it breaks TaskGroup, timeouts and shutdown, because they rely on it propagating. Put cleanup in finally blocks, such as releasing connections or closing streams, and keep that cleanup brief. It runs during cancellation, and any long await in it delays the cancel.

Cancellation can interrupt a coroutine between two writes, after a payment call but before the database update. Design those steps to be idempotent (for example, by using idempotency keys) or pair each with a compensating action.

Timeouts

On 3.11+, async with asyncio.timeout(...) puts one deadline around a whole block of awaits. asyncio.wait_for wraps a single awaitable and also works on older versions. Both raise TimeoutError on 3.11+.

Bounding gather

asyncio.gather over an unbounded list starts every coroutine at once and can exhaust pool slots, file descriptors and upstream rate limits. Acquire an asyncio.Semaphore inside each coroutine to cap how many run at the same time. With return_exceptions=True, failures come back as values in the results list and go unnoticed unless you check each one. With the default False, the first exception is raised, but the other awaitables keep running.

Illustrative estimate (assumed numbers, not a benchmark): wall time ≈ ceil(N / limit) × per-call latency. With N = 500 calls at 200 ms each, a limit of 10 takes 50 × 0.2 = 10 s, and a limit of 50 takes about 2 s. A limit set above your pool size gives no speedup, because the extra calls just wait for a connection. A limit set above the upstream's rate limit causes 429 errors and retries.

Mixing sync and async code and shutting down gracefully

Crossing the sync/async boundary

asyncio.run() creates a new event loop and closes it on exit, so it raises RuntimeError when a loop is already running in the current thread. If a sync helper inside your API calls asyncio.run() to reuse an async client, it fails there. Make the helper async and await it.

The reverse case needs a different tool. A worker thread, such as one started by asyncio.to_thread or a thread a third-party callback library owns, cannot await. It should call asyncio.run_coroutine_threadsafe(coro, loop), which schedules the coroutine on the loop's thread and returns a concurrent.futures.Future that the thread can wait on. Capture loop on the loop thread, typically with asyncio.get_running_loop(), before you hand work to the thread.

asyncio.Lock, asyncio.Queue and asyncio.Event are not thread-safe. They assume every caller runs on the loop thread. A put_nowait from another thread can corrupt state or never wake the waiter. Route such calls through loop.call_soon_threadsafe.

Shutting down in order

On Unix, register SIGTERM and SIGINT with loop.add_signal_handler so the handler runs on the loop rather than interrupting arbitrary code. When a signal arrives:

1. Stop accepting new work by closing the listener or marking the readiness check as failing.
2. Wait for in-flight tasks with asyncio.wait(..., timeout=deadline).
3. Cancel whatever remains, then await those tasks so their cleanup code runs.
4. Explicitly close HTTP clients, database pools and producers.

If you skip step 3, the interpreter garbage-collects pending tasks and logs "Task was destroyed but it is pending!". If you skip step 4, you get unclosed-connection warnings, and peers may see connections reset mid-response.

When to avoid asyncio, alternatives and impact on engineering cost

asyncio pays off when a service spends most of each request waiting on the network and has to hold many requests open at once. If your handlers mostly compute, or traffic is modest, a synchronous framework with more worker processes is often simpler and just as fast. Processes use every core, and a blocking call stalls only one worker, not the whole server.

Illustrative estimate (assumed numbers, not a benchmark). By Little's law, the number of requests in flight equals the request rate multiplied by latency. With an assumed 200 req/s and 0.5 s of mostly-I/O latency, 100 requests are in flight. Sync workers then need about 100 processes. At an assumed 60 MB each, that is 100 × 60 MB = 6 GB. One event loop per core avoids most of that cost. If the result is 5 in flight, the difference is negligible.

Weigh engineering cost too. Every dependency on the request path needs a mature async driver. Stack traces span tasks, which makes debugging harder. Teams new to the model ship more latency bugs. Detecting blocking calls in CI, isolating them in executors, and owning every task makes those incidents rarer and faster to diagnose.

How OneEnv can help

Disclosure: OneEnv is the product behind this blog.

OneEnv (oneenv.io) is a workspace-based governance platform for the things services share — environment variables, database structure, API routes and validation rules. A change to a shared value is staged on a branch, reviewed once, and then every affected service explicitly approves or rejects it before it goes live, so you can see who a change touches before it ships instead of finding out in production.

Applied to python asyncio, this is where it helps:

Saves time

• Every service that inherits a shared value is listed in one place, so nobody searches repos and chats to find out who is affected.
• One review per branch replaces coordinating each team separately; each service still keeps its own approve/reject.
• Changing a shared variable is one edit and one review instead of editing every service's .env by hand.

Saves cost

• Fewer production incidents and rollbacks caused by a shared value, route or column changing under a consumer unannounced.
• Less duplicated configuration to maintain: one shared definition inherited by many services.
• A wrong or stale value is caught at review time by the services it affects, not by a failed deploy.
• OneEnv has a free plan (1 workspace, unlimited services and members, full governance), so trying it costs nothing.

These are qualitative benefits, not measured results — how much you save depends on how many services share the value.

Want to try it? Start at oneenv.io.