aiohttp

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

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

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