aiohttp.ClientSession connection pooling within aiohttp.web.Application

main.py

import aiohttp
from aiohttp import web
from http import HTTPStatus
from typing import AsyncGenerator
from dataclasses import dataclass


@dataclass
class Config:
    timeout: int
    pool_size: int


async def fetch(client: aiohttp.ClientSession) -> str:
    response = await client.get("http://example.com/")
    assert response.status == HTTPStatus.OK
    return await response.text()


async def view_page(request: web.Request) -> web.Response:
    html = await fetch(request.app["client"])
    text = f"Recieved {len(html)} characters."
    return web.Response(text=text)


async def client_session(app: web.Application) -> AsyncGenerator[None, None]:
    config: Config = app["config"]
    timeout = aiohttp.ClientTimeout(total=config.timeout)
    connector = aiohttp.TCPConnector(limit=config.pool_size)
    async with aiohttp.ClientSession(timeout=timeout, connector=connector) as client:
        app["client"] = client
        yield


def create_app() -> web.Application:
    app = web.Application()
    app.add_routes([web.get("/", view_page)])
    app["config"] = Config(timeout=10, pool_size=100)
    app.cleanup_ctx.append(client_session)
    return app


if __name__ == "__main__":
    app = create_app()
    web.run_app(app)