import time
from dataclasses import dataclass
from typing import Optional
import requests
RETRYABLE_STATUS = {429, 500, 502, 503, 504}
@dataclass
class FetchResult:
url: str
status: Optional[int]
body: Optional[str]
error: Optional[str] = None
@property
def ok(self) -> bool:
return self.error is None
class HttpFetcher:
def __init__(self, timeout=5.0, max_retries=3, backoff_base=0.5):
self.timeout = timeout
self.max_retries = max_retries
self.backoff_base = backoff_base
self.session = requests.Session()
def _should_retry(self, status):
return status in RETRYABLE_STATUS
def fetch(self, url):
last_error = None
for attempt in range(self.max_retries + 1):
try:
resp = self.session.get(url, timeout=self.timeout)
if self._should_retry(resp.status_code) and attempt < self.max_retries:
last_error = "http {}".format(resp.status_code)
time.sleep(self.backoff_base * (2 ** attempt))
continue
resp.raise_for_status()
return FetchResult(url, resp.status_code, resp.text)
except (requests.ConnectionError, requests.Timeout) as exc:
last_error = str(exc)
if attempt < self.max_retries:
time.sleep(self.backoff_base * (2 ** attempt))
continue
except requests.HTTPError as exc:
status = exc.response.status_code if exc.response else None
return FetchResult(url, status, None, error=str(exc))
return FetchResult(url, None, None, error=last_error or "unknown error")
def close(self):
self.session.close()
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Iterable, List, Tuple
from fetcher import HttpFetcher, FetchResult
def fetch_all(urls: Iterable[str], max_workers=10, overall_timeout=60.0) -> List[FetchResult]:
urls = list(urls)
fetcher = HttpFetcher()
results: List[FetchResult] = []
try:
with ThreadPoolExecutor(max_workers=max_workers) as pool:
future_to_url = {pool.submit(fetcher.fetch, url): url for url in urls}
for future in as_completed(future_to_url, timeout=overall_timeout):
url = future_to_url[future]
try:
results.append(future.result())
except Exception as exc: # thread crashed before returning a result
results.append(FetchResult(url, None, None, error=repr(exc)))
finally:
fetcher.close()
return results
def try_map(urls: Iterable[str], **kwargs) -> Tuple[List[FetchResult], List[FetchResult]]:
results = fetch_all(urls, **kwargs)
ok = [r for r in results if r.ok]
failed = [r for r in results if not r.ok]
return ok, failed
if __name__ == "__main__":
targets = ["https://example.com/{}".format(i) for i in range(50)]
good, bad = try_map(targets, max_workers=8)
print("ok={} failed={}".format(len(good), len(bad)))
for r in bad[:5]:
print(" {} -> {}".format(r.url, r.error))
This snippet shows a common real-world pattern: firing many independent HTTP requests at once, but with a hard ceiling on concurrency so a downstream service (or the local process) is not overwhelmed. The work is split across a reusable fetcher that wraps a single request with retries, and a fan-out coordinator that schedules those fetches across a bounded pool and collects results as they finish.
In fetcher.py, HttpFetcher owns a requests.Session so connection pooling and keep-alive are reused across calls rather than paying a TCP/TLS handshake per request. The fetch method performs a single URL fetch with a per-attempt timeout and retries on transient failures — connection errors, timeouts, and 429/5xx responses — using exponential backoff derived from backoff_base. Non-retryable HTTP errors (like a 404) raise immediately, because retrying them wastes time and hides bugs. The result is normalized into a small FetchResult dataclass so the caller never has to touch a raw exception; failures are captured as data with the final error string.
The _should_retry helper centralizes the retry policy, which keeps the loop in fetch readable and makes the policy easy to audit. Honoring Retry-After on a 429 would be a natural extension here.
In fan_out.py, fetch_all is the coordinator. It creates a ThreadPoolExecutor whose max_workers bounds concurrency — the key detail that makes this safe against thousands of URLs. Because I/O-bound work releases the GIL while waiting on sockets, threads (not processes) are the right tool. Submissions are tracked in a future_to_url dict so results can be re-associated with their source URL, and as_completed yields futures the moment each finishes, letting results stream in out of order instead of blocking on the slowest request. An overall_timeout guards the whole batch. try_map demonstrates the typical consumer split: partition into successes and failures for downstream handling.
The trade-off is throughput versus politeness — a larger pool finishes faster but risks tripping rate limits, so max_workers should be tuned to the target's capacity. This approach fits I/O-bound fan-out over external APIs; CPU-bound work would want a process pool instead.
Related snips
import os
import stat
for root, _dirs, files in os.walk('/etc'):
for name in files:
path = os.path.join(root, name)Python security audit script for exposed risky filesystem state
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
class Product(models.Model):
name = models.CharField(max_length=200)
slug = models.SlugField(blank=True)
price = models.DecimalField(max_digits=10, decimal_places=2)
cost = models.DecimalField(max_digits=10, decimal_places=2)
margin = models.DecimalField(max_digits=5, decimal_places=2, blank=True)Django model signals vs overriding save
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
Share this code
Here's the card — post it anywhere.
