Python asyncio Patterns

# CLAUDE.md — Python asyncio Patterns

## Fundamentals

- Async functions are `async def`. They return coroutines. A coroutine doesn't run until awaited or scheduled as a task.
- One event loop per thread. Use `asyncio.run(main())` as the program's entry point.
- Don't mix sync blocking calls into async code. They block the loop and stall every other task.

## Tasks

- `asyncio.create_task(coro)` schedules a coroutine to run concurrently. Hold the reference — orphaned tasks can be garbage-collected mid-flight.
- For groups of related tasks, use `asyncio.TaskGroup` (3.11+):
  ```python
  async with asyncio.TaskGroup() as tg:
      tg.create_task(fetch_a())
      tg.create_task(fetch_b())
  ```
  TaskGroup propagates exceptions and cancels siblings on first failure. Prefer it over `gather` for new code.
- `asyncio.gather(*aws)` for parallel awaits when you need a list of results. Set `return_exceptions=True` only if you handle them.

## Cancellation

- Cancellation is cooperative. A task is cancelled by raising `CancelledError` at the next await point.
- Always handle `asyncio.CancelledError` in long-running tasks if you have cleanup to do, then re-raise.
- `try/finally` for cleanup. `finally` blocks run during cancellation.
- Don't catch `CancelledError` to "ignore" cancellation. That breaks every shutdown path.

## Timeouts

- Use `asyncio.timeout(seconds)` (3.11+):
  ```python
  async with asyncio.timeout(5):
      await slow_call()
  ```
  Older code uses `asyncio.wait_for`. Both raise `TimeoutError` on expiry.
- Always wrap network and IPC calls in a timeout. Hang-forever bugs are unrecoverable.

## Sync ↔ async bridges

- Run blocking code in a thread: `await asyncio.to_thread(blocking_func, arg)`. Pool managed by the loop.
- Don't call `asyncio.run` inside an async function — it creates a new loop and conflicts with the running one.
- Library code that *might* run in async contexts: prefer async-friendly libraries (`httpx` over `requests`, `aiosqlite` over `sqlite3`).

## Structured concurrency

- Prefer `TaskGroup` over standalone tasks. Lifetimes are visible from one place.
- Don't fire-and-forget. If a coroutine needs to outlive its caller, log that decision and hold the task somewhere reachable.

## Don't

- Don't sleep with `time.sleep(n)` in async code. Use `await asyncio.sleep(n)`.
- Don't share a single `aiohttp.ClientSession` (or any pooled resource) across `asyncio.run()` calls — it's tied to a loop.
- Don't use `asyncio.run()` more than once in a process if you can help it.
- Don't test async code with `asyncio.run` inside the test body. Use `pytest-asyncio` with `@pytest.mark.asyncio`.