GPT-5.6 Sol × async-fetch

6.3

Async concurrency patterns + retry/timeout handling

correctness 9.0quality 8.0documentation 2.018006ms

$ cat async-fetch.prompt — what the model was asked

Write a Python async function `fetch_all(urls: list[str], timeout: float = 5.0, retries: int = 2) -> list[dict]` that fetches all URLs concurrently using `aiohttp` and returns a list of result dicts.

Each result dict must have:
  - `url`: the original URL
  - `status`: HTTP status code (int), or None on error
  - `body`: response text (str), or None on error
  - `error`: error message string, or None on success

Requirements:
1. All URLs must be fetched concurrently (not sequentially)
2. Each fetch must respect the `timeout` (per-request, not total)
3. Retry up to `retries` times on network errors or 5xx responses (exponential backoff: 0.5s, 1s, ...)
4. Never raise — errors go into the result dict
5. Type-hint everything; add a docstring
6. Show a `__main__` block that fetches 3 URLs and prints results

run 1 of 1

6.3

correctness 9.0quality 8.0documentation 2.018006ms

# judge: Concurrency, per-attempt timeouts, retry targeting, exponential backoff, shared session use, and runnable main are strong, but dict types are unparameterized, Optional is unused, and the docstring omits required parameter, units, result-key, and backoff documentation.

$ cat codex-cli/gpt-5.6-sol.async-fetch.r1.response

```python
import asyncio
from typing import Optional

import aiohttp


async def fetch_all(
    urls: list[str],
    timeout: float = 5.0,
    retries: int = 2,
) -> list[dict]:
    """Fetch URLs concurrently with per-request timeouts and retries."""

    async def fetch_one(session: aiohttp.ClientSession, url: str) -> dict:
        attempts = max(0, retries) + 1

        for attempt in range(attempts):
            try:
                async with session.get(
                    url,
                    timeout=aiohttp.ClientTimeout(total=timeout),
                ) as response:
                    body = await response.text()

                    if response.status < 500:
                        return {
                            "url": url,
                            "status": response.status,
                            "body": body,
                            "error": None,
                        }

                    if attempt == attempts - 1:
                        return {
                            "url": url,
                            "status": response.status,
                            "body": body,
                            "error": f"HTTP {response.status}",
                        }

            except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
                if attempt == attempts - 1:
                    return {
                        "url": url,
                        "status": None,
                        "body": None,
                        "error": str(exc) or type(exc).__name__,
                    }
            except Exception as exc:
                return {
                    "url": url,
                    "status": None,
                    "body": None,
                    "error": str(exc) or type(exc).__name__,
                }

            await asyncio.sleep(0.5 * (2**attempt))

        return {
            "url": url,
            "status": None,
            "body": None,
            "error": "Unknown error",
        }

    try:
        async with aiohttp.ClientSession() as session:
            tasks = [fetch_one(session, url) for url in urls]
            return await asyncio.gather(*tasks)
    except Exception as exc:
        error: str = str(exc) or type(exc).__name__
        return [
            {"url": url, "status": None, "body": None, "error": error}
            for url in urls
        ]


if __name__ == "__main__":
    async def main() -> None:
        urls: list[str] = [
            "https://example.com",
            "https://httpbin.org/status/500",
            "https://httpbin.org/get",
        ]
        results: list[dict] = await fetch_all(urls)
        for result in results:
            print(result)

    asyncio.run(main())
```