The Four Disciplines of Production Asyncio

Python’s asyncio is exceptionally powerful for I/O-bound web services, but it is notoriously unforgiving when developers treat it like multi-threaded or synchronous code. Subtle bugs like silent task disappearance, event-loop starvation, and leaky context state can run undetected for months until traffic spikes.

Here are four battle-tested principles that ensure asynchronous Python services remain stable under heavy production load.

---

1. Single Ownership of the Event Loop

The most common source of runtime bugs (RuntimeError: Event loop is closed or Attached to a different loop) stems from ambiguous loop lifecycles.

The Golden Rule: One single event loop per operating system process, managed exclusively at the entry point of the application:

# Bad: Instantiating async clients at module import time
client = httpx.AsyncClient()  # Bound to an unknown or non-existent loop!

# Good: Instantiating network clients strictly inside an active async context
async def main():
    async with httpx.AsyncClient() as client:
        await run_service(client)

if __name__ == "__main__":
    asyncio.run(main())

Never instantiate stateful network connection pools, database engines, or Redis clients as module-level global variables before the event loop is booted.

---

2. Strong References for Background Fire-and-Forget Tasks

When launching non-blocking background tasks via asyncio.create_task(), Python’s garbage collector retains only a weak reference to the underlying coroutine object. If you do not preserve a strong reference in Python memory, the garbage collector may destroy the task mid-execution!

Always maintain a dedicated task registry:

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

def spawn_background_task(coro) -> asyncio.Task:
    task = asyncio.create_task(coro)
    background_tasks.add(task)
    # Automatically discard the reference upon task termination
    task.add_done_callback(background_tasks.discard)
    return task

Furthermore, during graceful shutdown routines, use asyncio.shield() when completing critical finalizations (such as closing billing transactions or flushing audit logs):

# Ensures the critical cleanup operation is shielded from abrupt cancellation
await asyncio.shield(commit_pending_transactions())

---

3. Safe Context Propagation with contextvars

Never use global variables or thread-local storage (threading.local) to track request context (such as user IDs or trace IDs) across async functions. In an asyncio application, hundreds of coroutines interleave on the same thread, causing global state to become contaminated.

Use contextvars.ContextVar:

import contextvars

request_id_ctx: contextvars.ContextVar[str] = contextvars.ContextVar("request_id")

def set_request_context(request_id: str):
    return request_id_ctx.set(request_id)

def get_current_request_id() -> str:
    return request_id_ctx.get(default="unknown")

Context variables are scoped to the active execution context and are automatically copied to sub-tasks spawned via asyncio.create_task().

---

4. Detecting and Eliminating Event Loop Starvation

Because asyncio relies on cooperative multitasking, a single blocking call freezes the entire event loop for every user on that worker process.

Never call synchronous I/O (requests.get, time.sleep, synchronous database drivers) within async code.

How to Catch Blockers in Staging:

Enable slow callback diagnostics during development:

loop = asyncio.get_running_loop()
# Emits a warning log whenever any coroutine occupies the CPU for more than 100ms
loop.slow_callback_duration = 0.1

If you must execute CPU-heavy calculations (image transformations, cryptographic hashing, large JSON deserialization) or legacy synchronous I/O, offload it immediately to a thread pool:

result = await asyncio.to_thread(heavy_cpu_or_sync_call, data)