---
description: Async Python
alwaysApply: false
---

# Async Python

`asyncio` enables concurrent I/O on a single thread. Understand the event loop, coroutines, and task lifecycle.

## Coroutines and Tasks

```python
async def fetch_data(url: str) -> bytes:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.read()

# Tasks run concurrently on the event loop
async def fetch_all(urls: list[str]) -> list[bytes]:
    tasks = [asyncio.create_task(fetch_data(url)) for url in urls]
    return await asyncio.gather(*tasks)
```

## gather vs TaskGroup

```python
# gather — returns results in order
users, posts = await asyncio.gather(fetch_users(), fetch_posts())

# TaskGroup (3.11+) — structured concurrency, cancels all on first failure
async with asyncio.TaskGroup() as tg:
    user_task = tg.create_task(fetch_users())
    post_task = tg.create_task(fetch_posts())
users, posts = user_task.result(), post_task.result()
```

## Timeouts and Bounded Concurrency

```python
# Timeout (3.11+)
async with asyncio.timeout(5.0):
    data = await fetch_data(url)

# Semaphore for bounding concurrency
semaphore = asyncio.Semaphore(10)
async def bounded_fetch(url: str) -> Response:
    async with semaphore:
        return await fetch(url)
```

## Key Rules

- Use `run_in_executor` for blocking/CPU-bound work in async context
- Async generators (`async for`) for streaming large result sets
- Every spawned task needs error handling and a cancellation strategy

## Anti-Patterns

```python
# Never: Blocking calls in async code
async def bad():
    requests.get(url)  # Blocks event loop — use aiohttp
    time.sleep(1)      # Blocks event loop — use asyncio.sleep()

# Never: Fire-and-forget without error handling
asyncio.create_task(work())  # Who catches exceptions?

# Never: Unbounded task spawning — use semaphore
for item in million_items:
    asyncio.create_task(process(item))  # OOM risk
```