How to implement async Python that actually works in production

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.