Python Async Programming: Mastering asyncio for High-Performance Applications

asyncio enables massive IO-bound concurrency on a single thread. Here's how to use it correctly.

The Mental Model

One thread, cooperative multitasking, explicit yield with await. IO-bound workloads see 10-100x speedups; CPU-bound needs multiprocessing.

import asyncio, aiohttp

# BAD: blocks entire event loop
async def fetch_bad(url: str) -> str:
    import requests
    return requests.get(url).text  # Blocks all other coroutines

# GOOD: non-blocking
async def fetch_good(session: aiohttp.ClientSession, url: str) -> str:
    async with session.get(url) as response:
        return await response.text()

Concurrent Requests

async def check_all(urls: list) -> list:
    async with aiohttp.ClientSession() as session:
        tasks = [get_status(session, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

Semaphores (Rate Control)

async def scrape_urls(urls: list, max_concurrent: int = 20) -> list:
    sem = asyncio.Semaphore(max_concurrent)

    async def fetch(url):
        async with sem:
            async with aiohttp.ClientSession() as s:
                async with s.get(url) as resp:
                    return await resp.text()

    return await asyncio.gather(*[fetch(u) for u in urls], return_exceptions=True)

Producer-Consumer Queue

async def pipeline(urls: list) -> list:
    queue, results, N = asyncio.Queue(maxsize=100), [], 5

    async def producer():
        for url in urls:
            await queue.put(url)
        for _ in range(N):
            await queue.put(None)  # Poison pills

    async def consumer():
        async with aiohttp.ClientSession() as session:
            while True:
                url = await queue.get()
                if url is None:
                    queue.task_done()
                    break
                try:
                    async with session.get(url) as resp:
                        results.append({'url': url, 'len': len(await resp.text())})
                finally:
                    queue.task_done()

    await asyncio.gather(producer(), *[asyncio.create_task(consumer()) for _ in range(N)])
    return results

CPU-Bound Work with ProcessPoolExecutor

from concurrent.futures import ProcessPoolExecutor
import multiprocessing

def heavy_compute(data: list) -> dict:
    return {'result': sum(x**2 for x in data)}

async def run_heavy(data: list) -> dict:
    loop = asyncio.get_event_loop()
    with ProcessPoolExecutor(max_workers=multiprocessing.cpu_count()) as ex:
        return await loop.run_in_executor(ex, heavy_compute, data)

Error Handling with Retry

async def with_retry(coro_fn, max_attempts=3, base_delay=1.0):
    for attempt in range(max_attempts):
        try:
            return await coro_fn()
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
            if attempt < max_attempts - 1:
                await asyncio.sleep(base_delay * (2 ** attempt))
            else:
                raise

asyncio rewards developers who respect its limits: IO-bound work, controlled concurrency, and CPU work in executors.

→ Test your async API endpoints with the URL Parser tool.