src/slop_farmer/data/http.py

from __future__ import annotations

import http.client
import socket
import time
import urllib.error
import urllib.request
from collections.abc import Callable
from typing import Any

TRANSIENT_HTTP_ERRORS = (
    TimeoutError,
    socket.timeout,
    urllib.error.URLError,
    http.client.RemoteDisconnected,
    ConnectionResetError,
)


def urlopen_with_retry(
    request: urllib.request.Request | str,
    *,
    timeout: int,
    max_retries: int = 5,
    log: Callable[[str], None] | None = None,
    label: str | None = None,
    opener: Callable[..., Any] | None = None,
    sleep: Callable[[float], None] = time.sleep,
) -> Any:
    attempt = 0
    target = label or (request if isinstance(request, str) else request.full_url)
    opener = opener or urllib.request.urlopen
    while True:
        try:
            return opener(request, timeout=timeout)
        except urllib.error.HTTPError:
            raise
        except TRANSIENT_HTTP_ERRORS as exc:
            attempt += 1
            if attempt > max_retries:
                raise RuntimeError(
                    f"HTTP request failed after {max_retries} retries: {target} {exc}"
                ) from exc
            sleep_for = min(2**attempt, 30)
            if log is not None:
                log(
                    f"Transient network failure for {target} (attempt {attempt}/{max_retries}); retrying in {sleep_for}s"
                )
            sleep(sleep_for)