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,
pandasoperations) 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:
- Never block the event loop — audit every library you import
- Always set timeouts — at the HTTP client level and per-operation with
wait_for - Use
gatherwithreturn_exceptions=Truefor fault-tolerant fan-out - Offload CPU work to
ProcessPoolExecutor, never to a coroutine - Handle
CancelledErrorproperly — clean up and re-raise - 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.
