You rewrote your service with asyncio and it still blocks. Or worse: it runs fine locally and falls apart under load. Async Python has a steep learning curve not because the syntax is hard, but because the mistakes are silent — a single blocking call freezes your entire event loop, and Python won't tell you which one.

How asyncio actually works

Asyncio runs on a single OS thread with a single event loop. That loop picks up coroutines, runs them until they hit an await, then switches to the next. This means:

  • Blocking I/O (synchronous requests.get(), time.sleep()) freezes everything
  • CPU-intensive work (heavy computation, pandas operations) freezes everything
  • Only I/O-bound concurrent work benefits from async

If you're not doing I/O, asyncio will make your code slower, not faster. This isn't a bug — it's the architecture. Keep it in mind before rewriting anything.

The most common mistake: accidentally blocking

Here's what trips up most developers:

import asyncio import httpx import time async def fetch_data(url: str) -> dict: # WRONG: time.sleep blocks the event loop — nothing else runs for 1 second  time.sleep(1) # WRONG: requests is synchronous under the hood  # import requests; return requests.get(url).json()  # RIGHT: async sleep + async HTTP client  await asyncio.sleep(1) async with httpx.AsyncClient() as client: response = await client.get(url, timeout=10.0) return response.json() 

The fix is systematic: replace every blocking call with its async equivalent. time.sleep → asyncio.sleep. requests → httpx or aiohttp. File I/O → aiofiles. Database calls → asyncpg, aiosqlite, or SQLAlchemy with its async engine. If a library you depend on has no async variant, see the CPU-bound section below — the same pattern applies.

Running tasks concurrently

The whole point of async is doing multiple I/O operations at once. The two main tools are asyncio.gather and asyncio.TaskGroup (Python 3.11+):

import asyncio import httpx from typing import Any async def fetch(client: httpx.AsyncClient, url: str) -> dict[str, Any]: response = await client.get(url, timeout=10.0) response.raise_for_status() return response.json() async def fetch_all(urls: list[str]) -> list[dict]: async with httpx.AsyncClient() as client: tasks = [fetch(client, url) for url in urls] # return_exceptions=True: one failure doesn't cancel the rest  results = await asyncio.gather(*tasks, return_exceptions=True) data = [] for i, result in enumerate(results): if isinstance(result, Exception): print(f"Failed {urls[i]}: {result}") else: data.append(result) return data 

Without return_exceptions=True, the first exception propagates immediately and you lose all other results. Use it whenever partial success is acceptable.

For Python 3.11+, TaskGroup gives you structured concurrency with cleaner error semantics:

async def fetch_all_v2(urls: list[str]) -> list[dict]: async with httpx.AsyncClient() as client: async with asyncio.TaskGroup() as tg: tasks = [tg.create_task(fetch(client, url)) for url in urls] # If any task raises, all tasks are cancelled and the exception re-raises  return [t.result() for t in tasks] 

Use TaskGroup when you need all-or-nothing semantics. Use gather when partial results are acceptable.

CPU-bound work: offload to a process pool

For anything CPU-intensive inside an async service, don't block the event loop — run it in a separate process:

import asyncio from concurrent.futures import ProcessPoolExecutor import hashlib def cpu_intensive(data: bytes) -> str: # Runs in a worker process, not the event loop thread  return hashlib.sha256(data * 50000).hexdigest() async def process_batch(items: list[bytes]) -> list[str]: loop = asyncio.get_running_loop() with ProcessPoolExecutor(max_workers=4) as executor: tasks = [loop.run_in_executor(executor, cpu_intensive, item) for item in items] return await asyncio.gather(*tasks) 

For blocking legacy libraries that are thread-safe, use ThreadPoolExecutor instead — same pattern, lighter weight than processes. The key point is loop.run_in_executor: it bridges sync code into the async world without freezing the loop.

Production patterns that actually matter

Always set timeouts. An async service with no timeouts is a service waiting to hang under partial failures. Set timeouts at the client level and per-operation:

# Client-level defaults client = httpx.AsyncClient( timeout=httpx.Timeout(connect=2.0, read=10.0, write=5.0, pool=1.0) ) # Per-operation timeout try: result = await asyncio.wait_for(slow_operation(), timeout=5.0) except asyncio.TimeoutError: logger.warning("Operation exceeded 5s budget") return None 

Handle cancellation properly. When a request times out or a client disconnects, asyncio cancels your coroutine by raising CancelledError. Never swallow it:

async def do_work() -> None: try: await long_operation() except asyncio.CancelledError: await cleanup() # release locks, close connections  raise # always re-raise — cancellation must propagate 

Limit concurrency with semaphores. Firing 1,000 concurrent requests will exhaust connection pools and get you blocked:

sem = asyncio.Semaphore(50) async def fetch_limited(client: httpx.AsyncClient, url: str) -> dict: async with sem: return await fetch(client, url) 

These patterns — timeouts, cancellation handling, concurrency limits — are the same controls that appear on our security hardening checklists for API services. Resource exhaustion is both a reliability failure and a security failure.

Debugging: find what blocks your loop

Enable asyncio's built-in debug mode during development:

import asyncio asyncio.run(main(), debug=True) 

This logs a warning for any callback that takes more than 100ms without yielding. You can tighten it:

loop = asyncio.get_event_loop() loop.slow_callback_duration = 0.05 # warn at 50ms 

For production on Linux, uvloop is a drop-in replacement for the default event loop that runs 2–4× faster on I/O-heavy workloads:

pip install uvloop 
import uvloop asyncio.set_event_loop_policy(uvloop.EventLoopPolicy()) asyncio.run(main()) 

Two lines for a meaningful throughput improvement on most HTTP servers.

The takeaway

Async Python works in production when you treat it as a discipline:

  1. Never block the event loop — audit every library you import
  2. Always set timeouts — at the HTTP client level and per-operation with wait_for
  3. Use gather with return_exceptions=True for fault-tolerant fan-out
  4. Offload CPU work to ProcessPoolExecutor, never to a coroutine
  5. Handle CancelledError properly — clean up and re-raise
  6. Enable debug mode in development to catch blocking calls before they reach production

Async doesn't make code faster by default. It makes I/O concurrent. Everything else — resource limits, error handling, observability — still needs the same attention you'd give any other service.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.