Python HTTP JSON Requests: Timeouts, Bounded Retries, and Local Tests

python3 --version
mkdir http-json-retry-demo
cd http-json-retry-demo
curl -fS https://www.bobobk.com/downloads/http-json-retry/get_json.py -o get_json.py
curl -fS https://www.bobobk.com/downloads/http-json-retry/demo_server.py -o demo_server.py
python3 get_json.py --help
python3 demo_server.py --port 8765

export no_proxy=127.0.0.1,localhost
export NO_PROXY=127.0.0.1,localhost

"""Fetch a small HTTP JSON document with bounded GET retries. Python 3.10+."""
import argparse
import json
import math
import sys
import time
from urllib.error import HTTPError, URLError
from urllib.parse import urlsplit
from urllib.request import Request, urlopen

MAX_BYTES = 1024 * 1024
RETRY_STATUS = {502, 503, 504}


def get_json(url, *, timeout=3.0, attempts=3):
    parts = urlsplit(url)
    if parts.scheme not in {"http", "https"} or not parts.hostname:
        raise ValueError("URL must use http:// or https:// with a host")
    if not math.isfinite(timeout) or timeout <= 0:
        raise ValueError("timeout must be finite and greater than zero")
    if not isinstance(attempts, int) or not 1 <= attempts <= 10:
        raise ValueError("attempts must be between 1 and 10")
    for attempt in range(1, attempts + 1):
        request = Request(url, headers={"Accept": "application/json"}, method="GET")
        try:
            with urlopen(request, timeout=timeout) as response:
                if response.status != 200:
                    raise ValueError(f"expected HTTP 200, got {response.status}")
                body = response.read(MAX_BYTES + 1)
                if len(body) > MAX_BYTES:
                    raise ValueError("response exceeds 1 MiB")
                return json.loads(body.decode("utf-8"))
        except HTTPError as exc:
            status = exc.code
            retry_after = exc.headers.get("Retry-After")
            exc.close()
            if status not in RETRY_STATUS or attempt == attempts:
                raise
            if retry_after is not None:
                raise ValueError(f"HTTP {status} with Retry-After; defer this job") from exc
            reason = f"HTTP {status}"
        except URLError as exc:
            if not isinstance(exc.reason, TimeoutError) or attempt == attempts:
                raise
            reason = "timeout"
        except TimeoutError:
            if attempt == attempts:
                raise
            reason = "timeout"
        delay = min(0.5 * 2 ** (attempt - 1), 8.0)
        print(
            f"RETRY attempt={attempt}/{attempts} reason={reason} delay={delay:.1f}s",
            file=sys.stderr,
            flush=True,
        )
        time.sleep(delay)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("url")
    parser.add_argument("--timeout", type=float, default=3.0, help="blocking timeout in seconds")
    parser.add_argument("--attempts", type=int, default=3, help="total attempts, including first")
    args = parser.parse_args()
    try:
        result = get_json(args.url, timeout=args.timeout, attempts=args.attempts)
    except (OSError, ValueError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

python3 get_json.py http://127.0.0.1:8765/flaky --timeout 1 --attempts 3

RETRY attempt=1/3 reason=HTTP 503 delay=0.5s
RETRY attempt=2/3 reason=HTTP 503 delay=1.0s
{
  "ok": true,
  "items": 3
}

python3 get_json.py http://127.0.0.1:8765/ok > result.json
python3 -c 'import json; data=json.load(open("result.json", encoding="utf-8")); assert data == {"ok": True, "items": 3}; print("PASS: expected data")'

python3 get_json.py http://127.0.0.1:8765/missing
python3 get_json.py http://127.0.0.1:8765/bad
python3 get_json.py http://127.0.0.1:8765/always503 --attempts 3
python3 get_json.py http://127.0.0.1:8765/slow --timeout 0.2 --attempts 2

"""Local HTTP failure fixture; never expose this as a production service."""
import argparse
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import threading
import time


class Handler(BaseHTTPRequestHandler):
    flaky_count = 0
    lock = threading.Lock()

    def do_GET(self):
        status, body = 200, b'{"ok": true, "items": 3}'
        if self.path == "/flaky":
            with Handler.lock:
                Handler.flaky_count += 1
                count = Handler.flaky_count
            if count <= 2:
                status, body = 503, b'{"error": "temporary"}'
        elif self.path == "/always503":
            status, body = 503, b'{"error": "unavailable"}'
        elif self.path == "/slow":
            time.sleep(2)
        elif self.path == "/bad":
            body = b'not json'
        elif self.path != "/ok":
            status, body = 404, b'{"error": "missing"}'
        try:
            self.send_response(status)
            self.send_header("Content-Type", "application/json; charset=utf-8")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
        except (BrokenPipeError, ConnectionResetError):
            pass  # The timeout demo can close the client socket first.

    def log_message(self, format, *args):
        pass


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--port", type=int, default=8765)
    args = parser.parse_args()
    with ThreadingHTTPServer(("127.0.0.1", args.port), Handler) as server:
        print(f"Serving http://127.0.0.1:{args.port}", flush=True)
        try:
            server.serve_forever()
        except KeyboardInterrupt:
            pass