Python Async

Python Async Patterns

Core Rules

• Never block the event loop: no time.sleep(), requests.get(), open() (without aiofiles) in async def
• Always await coroutines: forgetting await returns a coroutine object, not the result
• Use asyncio.TaskGroup (3.11+) for structured concurrency instead of asyncio.gather()
• Prefer anyio for library code to support both asyncio and trio

Task Groups (Structured Concurrency)

import asyncio

# Good: structured concurrency with TaskGroup (3.11+)
async def fetch_all(urls: list[str]) -> list[dict]:
    results = []
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch(url)) for url in urls]
    return [t.result() for t in tasks]

# For 3.9/3.10: use asyncio.gather with return_exceptions
async def fetch_all_compat(urls: list[str]) -> list[dict | Exception]:
    return await asyncio.gather(*[fetch(url) for url in urls], return_exceptions=True)

Semaphores for Rate Limiting

async def fetch_with_limit(urls: list[str], max_concurrent: int = 10) -> list[dict]:
    sem = asyncio.Semaphore(max_concurrent)
    
    async def bounded_fetch(url: str) -> dict:
        async with sem:
            return await fetch(url)
    
    return await asyncio.gather(*[bounded_fetch(url) for url in urls])

Async Context Managers

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # startup
    await db_pool.start()
    await cache.connect()
    yield
    # shutdown
    await db_pool.close()
    await cache.disconnect()

Background Tasks with Error Capture

import asyncio
import logging

logger = logging.getLogger(__name__)

def create_task_with_logging(coro, *, name: str | None = None) -> asyncio.Task:
    task = asyncio.create_task(coro, name=name)
    task.add_done_callback(_log_task_exception)
    return task

def _log_task_exception(task: asyncio.Task) -> None:
    if not task.cancelled() and (exc := task.exception()):
        logger.exception("Background task %s failed", task.get_name(), exc_info=exc)

Queue-Based Worker Pool

async def worker_pool(jobs: list[Job], num_workers: int = 5) -> list[Result]:
    queue: asyncio.Queue[Job] = asyncio.Queue()
    results: list[Result] = []
    
    for job in jobs:
        await queue.put(job)
    
    async def worker() -> None:
        while True:
            try:
                job = queue.get_nowait()
            except asyncio.QueueEmpty:
                break
            result = await process(job)
            results.append(result)
            queue.task_done()
    
    await asyncio.gather(*[worker() for _ in range(num_workers)])
    return results

Anti-Patterns

# Bad: blocking I/O in async function
async def bad_read_file(path: str) -> str:
    return open(path).read()  # blocks event loop

# Good: use aiofiles
import aiofiles
async def good_read_file(path: str) -> str:
    async with aiofiles.open(path) as f:
        return await f.read()

# Bad: deprecated event loop access
loop = asyncio.get_event_loop()  # deprecated in 3.10+

# Good: use asyncio.run() at entry point
if __name__ == "__main__":
    asyncio.run(main())