Python asyncio Advanced Patterns: Concurrency Without Threads

asyncio in Production

Python's asyncio excels at I/O-bound concurrency. Here are the patterns that matter in real applications.

Task Groups (Python 3.11+)

import asyncio

async def fetch_data(url: str) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()

async def main():
    urls = [
        "https://api.example.com/users",
        "https://api.example.com/products",
        "https://api.example.com/orders",
    ]
    
    # TaskGroup cancels all on first failure
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch_data(url)) for url in urls]
    
    results = [t.result() for t in tasks]
    return results

Semaphores for Rate Limiting

async def fetch_with_limit(url: str, semaphore: asyncio.Semaphore) -> dict:
    async with semaphore:
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                return await response.json()

async def fetch_many(urls: list[str], max_concurrent: int = 10):
    semaphore = asyncio.Semaphore(max_concurrent)
    tasks = [fetch_with_limit(url, semaphore) for url in urls]
    return await asyncio.gather(*tasks, return_exceptions=True)

Producer-Consumer with Queues

import asyncio
from asyncio import Queue

async def producer(queue: Queue, items: list):
    for item in items:
        await queue.put(item)
        await asyncio.sleep(0.01)  # Simulate production delay
    
    # Signal completion
    await queue.put(None)

async def consumer(queue: Queue, worker_id: int, results: list):
    while True:
        item = await queue.get()
        if item is None:
            queue.task_done()
            await queue.put(None)  # Pass signal to next consumer
            break
        
        result = await process_item(item)
        results.append(result)
        queue.task_done()
        print(f"Worker {worker_id}: processed {item}")

async def pipeline(items: list, num_workers: int = 4):
    queue = Queue(maxsize=100)  # Backpressure!
    results = []
    
    consumers = [
        asyncio.create_task(consumer(queue, i, results))
        for i in range(num_workers)
    ]
    
    await producer(queue, items)
    await asyncio.gather(*consumers)
    
    return results

Timeout and Cancellation

import asyncio

async def fetch_with_timeout(url: str, timeout: float = 5.0) -> dict | None:
    try:
        async with asyncio.timeout(timeout):  # Python 3.11+
            return await fetch_data(url)
    except TimeoutError:
        print(f"Timeout fetching {url}")
        return None

# Handle cancellation gracefully
async def long_running_task():
    try:
        for i in range(100):
            await do_work(i)
            await asyncio.sleep(0.1)
    except asyncio.CancelledError:
        print("Task cancelled — cleaning up")
        await cleanup()
        raise  # Re-raise to propagate

# Cancel a task
task = asyncio.create_task(long_running_task())
await asyncio.sleep(1)
task.cancel()
try:
    await task
except asyncio.CancelledError:
    print("Task was cancelled")

Connection Pool Pattern

import asyncio
from asyncio import Queue
import asyncpg

class AsyncConnectionPool:
    def __init__(self, dsn: str, min_size: int = 5, max_size: int = 20):
        self.dsn = dsn
        self.min_size = min_size
        self.max_size = max_size
        self._pool: asyncpg.Pool | None = None
    
    async def __aenter__(self):
        self._pool = await asyncpg.create_pool(
            self.dsn,
            min_size=self.min_size,
            max_size=self.max_size,
        )
        return self
    
    async def __aexit__(self, *args):
        await self._pool.close()
    
    async def execute(self, query: str, *args):
        async with self._pool.acquire() as conn:
            return await conn.execute(query, *args)
    
    async def fetch(self, query: str, *args) -> list:
        async with self._pool.acquire() as conn:
            return await conn.fetch(query, *args)

Running Sync Code in Threads

import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=10)

async def run_sync(func, *args):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(executor, func, *args)

# Example: blocking PIL image processing
async def process_image(image_path: str):
    def _process(path: str):
        from PIL import Image
        img = Image.open(path)
        img = img.resize((800, 600))
        img.save(path.replace('.jpg', '_resized.jpg'))
    
    await run_sync(_process, image_path)

Structured Concurrency Pattern

from contextlib import asynccontextmanager
from typing import AsyncIterator

@asynccontextmanager
async def managed_tasks(*coros) -> AsyncIterator[list]:
    tasks = [asyncio.create_task(coro) for coro in coros]
    try:
        yield tasks
        await asyncio.gather(*tasks)
    except Exception:
        # Cancel all tasks on error
        for task in tasks:
            task.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)
        raise

async def main():
    async with managed_tasks(
        fetch_users(),
        fetch_products(),
        warm_cache(),
    ) as tasks:
        pass  # All tasks complete or all are cancelled

Async Context Variables

from contextvars import ContextVar

request_id: ContextVar[str] = ContextVar('request_id', default='')

async def handle_request(req_id: str):
    token = request_id.set(req_id)
    try:
        await process_request()
    finally:
        request_id.reset(token)

async def process_request():
    rid = request_id.get()
    print(f"Processing request {rid}")
    # Works correctly even with concurrent requests

Performance Tips

• Use asyncio.gather() for parallel independent tasks
• Use asyncio.Queue for backpressure in pipelines
• Avoid time.sleep() — use asyncio.sleep()
• Profile with aiomonitor or py-spy
• Set PYTHONASYNCIODEBUG=1 in development to catch issues