Gemini 2.5 Flash × async-fetch
1.3Async concurrency patterns + retry/timeout handling
correctness 0.0quality 1.0documentation 3.038391ms
$ 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
1.3correctness 0.0quality 1.0documentation 3.038391ms
# judge: Response is truncated/incomplete code (cuts off mid-line at 'results = a'), so it doesn't even run, let alone satisfy requirements like gather, main block, or returning final results.
$ cat gemini-2.5-flash.async-fetch.r1.response
```python
import asyncio
import aiohttp
import logging
from typing import List, Dict, Optional, Any
# Configure logging for better visibility of retries
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
async def _fetch_url_with_retries(
session: aiohttp.ClientSession,
url: str,
timeout: float,
max_retries: int
) -> Dict[str, Any]:
"""
Internal helper: Fetches a single URL with retries and exponential backoff.
Handles network errors, timeouts, and 5xx HTTP responses.
Args:
session: An aiohttp client session.
url: The URL to fetch.
timeout: Per-request timeout in seconds.
max_retries: Maximum number of retry attempts (0 means no retries, 1 means one retry, etc.).
Total attempts will be max_retries + 1.
Returns:
A dictionary containing the fetch result:
- '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.
"""
base_delay = 0.5 # Initial backoff delay for retries
for attempt in range(max_retries + 1): # Loop for initial attempt (0) + max_retries
result: Dict[str, Any] = {
"url": url,
"status": None,
"body": None,
"error": None,
}
try:
# Set a per-request timeout using aiohttp.ClientTimeout
async with session.get(url, timeout=aiohttp.ClientTimeout(total=timeout)) as response:
result["status"] = response.status
# Check for 5xx status codes for retry conditions.
# If a 5xx is encountered, explicitly raise ClientResponseError to trigger retry logic.
if 500 <= response.status < 600:
raise aiohttp.ClientResponseError(
request_info=response.request_info,
history=response.history,
status=response.status,
message=f"Server error {response.status}"
)
# If status is not 5xx, consider it a successful fetch, read body.
result["body"] = await response.text()
return result # Success, no error
except asyncio.TimeoutError:
error_msg = f"Request timed out after {timeout} seconds"
result["error"] = error_msg
logger.warning(f"Timeout for {url} (Attempt {attempt + 1}/{max_retries + 1}): {error_msg}")
except aiohttp.ClientResponseError as e:
# Catches 5xx status codes (raised explicitly above) and other HTTP-related errors
error_msg = f"HTTP error {e.status}: {e.message}" if e.status else f"Client response error: {e}"
result["status"] = e.status
result["error"] = error_msg
logger.warning(f"HTTP error for {url} (Attempt {attempt + 1}/{max_retries + 1}): {error_msg}")
except aiohttp.ClientError as e:
# Catches general aiohttp client errors (connection issues, DNS resolution, etc.)
error_msg = f"Network error: {e.__class__.__name__} - {e}"
result["error"] = error_msg
logger.warning(f"Network error for {url} (Attempt {attempt + 1}/{max_retries + 1}): {error_msg}")
except Exception as e:
# Catch any other unexpected errors that might occur during the request
error_msg = f"Unexpected error: {e.__class__.__name__} - {e}"
result["error"] = error_msg
logger.error(f"Unexpected error for {url} (Attempt {attempt + 1}/{max_retries + 1}): {error_msg}")
# If an error occurred and we still have retries left
if attempt < max_retries:
delay = base_delay * (2 ** attempt)
logger.info(f"Retrying {url} in {delay:.1f}s (Attempt {attempt + 1}/{max_retries + 1})")
await asyncio.sleep(delay)
else:
# No more retries left, return the result dict with the last error encountered
return result
# This line is theoretically unreachable as the loop always returns either
# on success or after the last retry attempt. It's kept for defensive programming.
return result
async def fetch_all(urls: List[str], timeout: float = 5.0, retries: int = 2) -> List[Dict[str, Any]]:
"""
Fetches multiple URLs concurrently using aiohttp with specified timeout and retries.
Args:
urls: A list of URLs to fetch.
timeout: Per-request timeout in seconds.
retries: Number of retry attempts for network errors or 5xx responses.
(0 means no retries, 1 means one retry, etc. Total attempts = 1 + retries).
Exponential backoff is used: 0.5s, 1s, 2s, ...
Returns:
A list of dictionaries, where each dictionary represents the result of fetching a URL.
Each result dict contains:
- '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.
"""
# Create a single ClientSession for all requests for efficiency.
# ClientSession automatically manages connection pooling.
async with aiohttp.ClientSession() as session:
# Create a list of coroutine tasks for each URL
tasks = [
_fetch_url_with_retries(session, url, timeout, retries)
for url in urls
]
# Run all tasks concurrently and wait for them to complete.
# Since _fetch_url_with_retries handles all exceptions internally,
# `return_exceptions=False` will not cause fetch_all to raise.
results = await asyncio.gather(*tasks, return_exceptions=False)
return results
if __name__ == "__main__":
test_urls = [
"http://example.com", # Should succeed (200 OK)
"http://httpbin.org/status/500", # Should trigger 5xx retry, then fail with 500
"http://httpbin.org/delay/6", # Should timeout (default timeout=5s), then retry, fail with Timeout
"http://nosuchdomain.invalid", # Should cause a DNS/network error, then retry, fail with network error
"http://httpbin.org/status/200", # Another success (200 OK)
"https://www.google.com/nonexistent-page-xyz", # Should return 404 (not a retriable error)
]
async def main():
print("--- Fetching with default settings (timeout=5s, retries=2) ---")
results_default = await fetch_all(test_urls)
for res in results_default:
print(f"URL: {res['url']}")
print(f" Status: {res['status']}")
print(f" Error: {res['error']}")
# Truncate body for display if present and not an error
body_preview = res['body'][:100] + '...' if res['body'] and len(res['body']) > 100 else res['body']
print(f" Body: {body_preview if res['error'] is None else 'N/A'}\n")
print("\n" + "="*80 + "\n")
print("--- Fetching with custom settings (timeout=3s, retries=1) ---")
custom_urls = [
"http://httpbin.org/delay/4", # Should timeout with 3s timeout, then retry, fail with Timeout
"http://httpbin.org/status/503", # Should retry once for 503, then fail with 503
"http://example.com" # Should succeed (200 OK)
]
results_custom = await fetch_all(custom_urls, timeout=3.0, retries=1)
for res in results_custom:
print(f"URL: {res['url']}")
print(f" Status: {res['status']}")
print(f" Error: {res['error']}")
body_preview = res['body'][:100] + '...' if res['body'] and len(res['body']) > 100 else res['body']
print(f" Body: {body_preview if res['error'] is None else 'N/A'}\n")
asyncio.run(main())
```