Fetch URLs concurrently with aiohttp

import asyncio

import aiohttp


async def fetch(session: aiohttp.ClientSession, url: str) -> tuple[str, int, int]:
    """Return (url, status, bytes); HTTP errors are reported, not raised."""
    try:
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
            body = await resp.read()
            return url, resp.status, len(body)
    except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
        print(f"{url}: {exc!r}")
        return url, 0, 0


async def fetch_all(urls: list[str], limit: int = 5) -> list[tuple[str, int, int]]:
    """Fetch every URL with at most `limit` requests in flight."""
    semaphore = asyncio.Semaphore(limit)
    connector = aiohttp.TCPConnector(limit=limit)

    async def guarded(session: aiohttp.ClientSession, url: str):
        async with semaphore:
            return await fetch(session, url)

    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [guarded(session, url) for url in urls]
        return await asyncio.gather(*tasks)


if __name__ == "__main__":
    urls = ["https://example.com", "https://www.python.org"]
    for result in asyncio.run(fetch_all(urls)):
        print(result)