aiohttp

The skill Copy

# `aiohttp` Best Practices

`aiohttp` is our go-to for high-performance asynchronous HTTP client operations. These rules ensure our `aiohttp` code is consistent, efficient, and resilient.

## 1. Manage `ClientSession` Lifecycle Properly

Always use a single, long-lived `ClientSession` per application or logical component. This leverages connection pooling and keep-alive, drastically reducing overhead. Always use it as an `async` context manager to ensure connections are closed cleanly.

❌ **BAD: Creating a new session for every request**
```python
import aiohttp
import asyncio

async def fetch_data(url: str) -> str:
    async with aiohttp.ClientSession() as session: # Session created and closed per call
        async with session.get(url) as response:
            return await response.text()

async def main():
    await fetch_data("http://example.com/api/v1/data")
    await fetch_data("http://example.com/api/v1/status") # New session, new connection
```

✅ **GOOD: Single, long-lived `ClientSession`**
```python
import aiohttp
import asyncio

# Create the session once, typically at application startup
# or pass it down through dependencies.
_GLOBAL_HTTP_SESSION: aiohttp.ClientSession | None = None

async def get_http_session() -> aiohttp.ClientSession:
    global _GLOBAL_HTTP_SESSION
    if _GLOBAL_HTTP_SESSION is None or _GLOBAL_HTTP_SESSION.closed:
        _GLOBAL_HTTP_SESSION = aiohttp.ClientSession()
    return _GLOBAL_HTTP_SESSION

async def close_http_session():
    global _GLOBAL_HTTP_SESSION
    if _GLOBAL_HTTP_SESSION and not _GLOBAL_HTTP_SESSION.closed:
        await _GLOBAL_HTTP_SESSION.close()
        _GLOBAL_HTTP_SESSION = None

async def fetch_data(session: aiohttp.ClientSession, url: str) -> str:
    async with session.get(url) as response:
        response.raise_for_status() # Always raise for non-2xx statuses
        return await response.text()

async def main():
    session = await get_http_session()
    try:
        data = await fetch_data(session, "http://example.com/api/v1/data")
        status = await fetch_data(session, "http://example.com/api/v1/status")
        print(f"Data: {data[:20]}..., Status: {status[:20]}...")
    finally:
        await close_http_session() # Ensure session is closed on app shutdown
```

## 2. Install All Speedups

Always install `aiohttp` with its `[speedups]` extra. This includes `aiodns` for faster DNS resolution and `backports.zstd` (for Python < 3.14) for modern compression, significantly boosting performance.

```bash
pip install aiohttp[speedups]
```

## 3. Implement Robust Timeouts and Error Handling

Explicitly define timeouts using `aiohttp.ClientTimeout` and enable `raise_for_status=True` for automatic HTTP error detection. Wrap requests in `try/except aiohttp.ClientError` to handle network and HTTP-level issues gracefully.

❌ **BAD: No timeouts, generic exception handling**
```python
async def fetch_unreliable(session: aiohttp.ClientSession, url: str) -> str:
    # This request could hang indefinitely
    async with session.get(url) as response:
        # 4xx/5xx responses won't raise an error, requiring manual check
        return await response.text()
```

✅ **GOOD: Explicit timeouts, `raise_for_status`, and specific error handling**
```python
from aiohttp import ClientSession, ClientTimeout, ClientError, ClientResponseError
import asyncio

async def fetch_reliable(session: ClientSession, url: str) -> str:
    timeout = ClientTimeout(total=5, connect=1, sock_read=3) # 5s total, 1s connect, 3s read
    try:
        async with session.get(url, timeout=timeout, raise_for_status=True) as response:
            return await response.text()
    except ClientResponseError as e:
        print(f"HTTP error for {url}: {e.status} - {e.message}")
        raise
    except ClientError as e:
        print(f"Network or client error for {url}: {e}")
        raise
    except asyncio.TimeoutError:
        print(f"Request to {url} timed out.")
        raise
```

## 4. Use Typed, Reusable Request Helpers

Encapsulate common request patterns in small, type-hinted `async` functions. This improves readability, maintainability, and enables better IDE support.

❌ **BAD: Repeated request logic and untyped functions**
```python
async def get_user_data(session, user_id):
    resp = await session.get(f"http://api.example.com/users/{user_id}")
    return await resp.json()

async def create_user(session, user_data):
    resp = await session.post("http://api.example.com/users", json=user_data)
    return await resp.json()
```

✅ **GOOD: Type-hinted, reusable functions with `json=` shortcut**
```python
from typing import Any, Dict
from aiohttp import ClientSession

async def fetch_json(session: ClientSession, url: str) -> Dict[str, Any]:
    """Fetches JSON from a URL, raises for status, and returns parsed data."""
    async with session.get(url, raise_for_status=True) as response:
        return await response.json()

async def post_json(session: ClientSession, url: str, payload: Dict[str, Any]) -> Dict[str, Any]:
    """Posts JSON payload to a URL, raises for status, and returns parsed response."""
    async with session.post(url, json=payload, raise_for_status=True) as response:
        return await response.json()

async def get_user_data(session: ClientSession, user_id: str) -> Dict[str, Any]:
    return await fetch_json(session, f"http://api.example.com/users/{user_id}")

async def create_user(session: ClientSession, user_data: Dict[str, Any]) -> Dict[str, Any]:
    return await post_json(session, "http://api.example.com/users", user_data)
```

## 5. Prefer Built-in Authentication and Session Headers

Use the `auth` parameter for basic authentication or `DigestAuthMiddleware` for digest auth. For other authentication schemes or default headers, set them when initializing `ClientSession` or update `session.headers`. Avoid manual `Authorization` header manipulation where built-in options exist.

❌ **BAD: Manually constructing `Authorization` headers**
```python
import base64
async def fetch_with_basic_auth(session: ClientSession, url: str, user: str, password: str):
    auth_str = base64.b64encode(f"{user}:{password}".encode()).decode()
    headers = {"Authorization": f"Basic {auth_str}"}
    async with session.get(url, headers=headers) as response:
        return await response.text()
```

✅ **GOOD: Using `auth` parameter for basic auth**
```python
from aiohttp import ClientSession, BasicAuth

async def fetch_with_basic_auth(session: ClientSession, url: str, user: str, password: str):
    auth = BasicAuth(login=user, password=password)
    async with session.get(url, auth=auth) as response:
        response.raise_for_status()
        return await response.text()

# For session-wide default headers (e.g., Bearer tokens)
async def create_auth_session(token: str) -> ClientSession:
    headers = {"Authorization": f"Bearer {token}"}
    return ClientSession(headers=headers)
```

## 6. Leverage Client Middleware for Cross-Cutting Concerns

For concerns like tracing, retries, or rate limiting that apply across many requests, implement client-side middleware. This keeps your request logic clean and separates concerns effectively.

```python
from aiohttp import ClientSession, ClientRequest, ClientResponse
from aiohttp.typedefs import ClientHandler
import logging

logging.basicConfig(level=logging.INFO)

async def logging_middleware(req: ClientRequest, handler: ClientHandler) -> ClientResponse:
    """Logs request and response details."""
    logging.info(f"Requesting: {req.method} {req.url}")
    response = await handler(req)
    logging.info(f"Received: {response.status} for {req.method} {req.url}")
    return response

async def main_with_middleware():
    async with ClientSession(middlewares=(logging_middleware,)) as session:
        await fetch_json(session, "http://httpbin.org/get")
        await post_json(session, "http://httpbin.org/post", {"key": "value"})

# Example usage with the `fetch_json` and `post_json` helpers from above.
```

How to install it Copy

1. Open the tool and start a new conversation, project or template.
2. Paste the text below wherever the tool keeps standing instructions
   — usually called system prompt, instructions, or persona.
3. Save it so it applies to every run rather than a single message.

Did it work for you?

Where it came from