aiohttp — Streaming and Server Framework

Beyond client basics: streaming uploads/downloads, WebSockets, and aiohttp as a server framework. The server side is less common in 2025 (FastAPI dominates) but you’ll see it in legacy code.

Streaming downloads

For large responses, don’t await resp.read() (loads into memory). Stream:

async with session.get(url) as resp:
    async for chunk in resp.content.iter_chunked(64 * 1024):
        await write_to_file(chunk)

iter_chunked yields chunks as they arrive. Memory stays flat regardless of response size.

For line-by-line (NDJSON, CSV):

async for line in resp.content:
    process(line)        # yields bytes per line

For raw streaming without buffering:

async for chunk in resp.content.iter_any():
    process(chunk)       # whatever the socket gave us

Streaming uploads

async def file_chunks(path):
    async with aiofiles.open(path, "rb") as f:
        while chunk := await f.read(64 * 1024):
            yield chunk

async with session.post(url, data=file_chunks(path)) as resp:
    ...

Passing an async generator as data streams the body. No loading the whole file into memory.

Or with a file object (sync, gets streamed under the hood):

with open(path, "rb") as f:
    async with session.post(url, data=f) as resp:
        ...

For multipart:

data = aiohttp.FormData()
data.add_field("file", open(path, "rb"), filename=os.path.basename(path), content_type="application/octet-stream")
async with session.post(url, data=data) as resp:
    ...

WebSocket client

async with session.ws_connect("wss://api.example.com/socket") as ws:
    await ws.send_str(json.dumps({"action": "subscribe", "channel": "orders"}))
    async for msg in ws:
        if msg.type == aiohttp.WSMsgType.TEXT:
            process(json.loads(msg.data))
        elif msg.type == aiohttp.WSMsgType.ERROR:
            break

Long-lived bidirectional connection. Reconnect logic is your job:

while not stopping:
    try:
        async with session.ws_connect(url) as ws:
            ...
    except (aiohttp.ClientError, asyncio.TimeoutError):
        await asyncio.sleep(backoff)

For production WebSocket clients, the websockets library is generally preferred. aiohttp’s WS support is fine but less feature-rich.

Server framework

aiohttp also includes a web framework:

from aiohttp import web

async def handler(request):
    name = request.match_info["name"]
    return web.json_response({"hello": name})

app = web.Application()
app.add_routes([web.get("/hello/{name}", handler)])

if __name__ == "__main__":
    web.run_app(app, host="0.0.0.0", port=8080)

Async out of the box. Predates FastAPI; older codebases often use it.

Middleware

@web.middleware
async def auth_middleware(request, handler):
    if "Authorization" not in request.headers:
        return web.json_response({"error": "unauth"}, status=401)
    return await handler(request)

app = web.Application(middlewares=[auth_middleware])

Closure pattern. Simpler than Django/Flask middleware classes.

Lifecycle hooks

async def startup(app):
    app["db"] = await asyncpg.create_pool(...)

async def cleanup(app):
    await app["db"].close()

app.on_startup.append(startup)
app.on_cleanup.append(cleanup)

Shared state via app[key] (dict-like). Available in handlers via request.app["db"].

Class-based views

class UserView(web.View):
    async def get(self):
        user_id = self.request.match_info["id"]
        return web.json_response({"id": user_id})

    async def post(self):
        body = await self.request.json()
        return web.json_response({"created": True})

app.add_routes([web.view("/users/{id}", UserView)])

Less common; most aiohttp code uses function-based handlers.

aiohttp server vs FastAPI

For new projects: FastAPI. For aiohttp legacy code: maintain it, migrate gradually if performance is fine.

Performance notes

• aiohttp server is HTTP/1.1 only (HTTP/2 in 4.0 beta). For HTTP/2, use FastAPI behind hypercorn or a Caddy/nginx front.
• No native async file I/O. Use aiofiles for non-blocking disk reads.
• Built-in CORS via aiohttp-cors extension (not in core).
• No built-in OpenAPI — use aiohttp-apispec or similar.

When to use aiohttp client (the more common case)

• Service-to-service HTTP from async Python.
• Large file transfers (streaming).
• WebSocket clients (though websockets lib is purer).
• Existing codebase already uses it — stay consistent.

For new client code, httpx is often the better default — supports sync+async, HTTP/2, cleaner API.

Common patterns

Concurrent requests with bounded concurrency

async def fetch_all(urls, max_concurrent=20):
    sem = asyncio.Semaphore(max_concurrent)
    async def one(url):
        async with sem:
            async with session.get(url) as resp:
                return await resp.json()
    return await asyncio.gather(*[one(u) for u in urls])

Don’t fire all N requests at once — overwhelm yourself or the downstream. Semaphore caps concurrency.

Periodic health pings

async def health_check():
    while True:
        try:
            async with session.get(HEALTH_URL, timeout=ClientTimeout(total=2)) as resp:
                if resp.status != 200:
                    log.warning("Unhealthy", status=resp.status)
        except (ClientError, asyncio.TimeoutError):
            log.warning("Health check failed")
        await asyncio.sleep(30)

Background task started in lifespan; cancelled on shutdown.

Server-Sent Events client

async with session.get(sse_url, headers={"Accept": "text/event-stream"}) as resp:
    async for line in resp.content:
        line = line.decode().strip()
        if line.startswith("data:"):
            handle_event(line[5:])

Same as a regular streaming download; SSE is just text/event-stream framing.

Interview angle

• “How do you stream a large file response without loading it into memory?” — async for chunk in resp.content.iter_chunked(N). Each chunk is yielded as it arrives; memory stays flat regardless of response size.
• “How do you stream an upload?” — pass an async generator as data. aiohttp pulls chunks lazily; no full-body buffering in memory.
• “WebSocket client in aiohttp vs websockets lib?” — aiohttp’s WS is fine for occasional use; websockets library is purer-async and more featureful. For a Python service whose primary job is WS, prefer websockets.
• “aiohttp server vs FastAPI?” — FastAPI for new code (Pydantic, auto OpenAPI, DI, ecosystem). aiohttp for legacy or when you need bare-metal minimal overhead. Both are async-first; performance is comparable.
• “How do you do CORS in aiohttp?” — aiohttp-cors extension. Not in core; install separately.
• “Bounded concurrency for 1000 outbound requests?” — asyncio.Semaphore(N) around each request, then asyncio.gather the lot. Without a semaphore, you fire 1000 requests simultaneously and overwhelm the downstream and your own pool.