# Python Async

> When to activate: asyncio, async/await, concurrent I/O, aiohttp, anyio, task groups, asyncio patterns

- Skill: `mattakushi432/python-async` (Agent Skill)
- Install (CLI): `npx skillmds@latest add mattakushi432/python-async`
- Raw SKILL.md: https://api.skillmd.com/api/skills/mattakushi432/python-async/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Productivity
- Author: Mattakushi432 (https://skillmd.com/u/mattakushi432)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/mattakushi432/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)
```python
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
```python
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
```python
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
```python
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
```python
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
```python
# 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())
```