evalstate/diffusers-pr-api

1from __future__ import annotations2 3import http.client4import socket5import time6import urllib.error7import urllib.request8from collections.abc import Callable9from typing import Any10 11TRANSIENT_HTTP_ERRORS = (12    TimeoutError,13    socket.timeout,14    urllib.error.URLError,15    http.client.RemoteDisconnected,16    ConnectionResetError,17)18 19 20def urlopen_with_retry(21    request: urllib.request.Request | str,22    *,23    timeout: int,24    max_retries: int = 5,25    log: Callable[[str], None] | None = None,26    label: str | None = None,27    opener: Callable[..., Any] | None = None,28    sleep: Callable[[float], None] = time.sleep,29) -> Any:30    attempt = 031    target = label or (request if isinstance(request, str) else request.full_url)32    opener = opener or urllib.request.urlopen33    while True:34        try:35            return opener(request, timeout=timeout)36        except urllib.error.HTTPError:37            raise38        except TRANSIENT_HTTP_ERRORS as exc:39            attempt += 140            if attempt > max_retries:41                raise RuntimeError(42                    f"HTTP request failed after {max_retries} retries: {target} {exc}"43                ) from exc44            sleep_for = min(2**attempt, 30)45            if log is not None:46                log(47                    f"Transient network failure for {target} (attempt {attempt}/{max_retries}); retrying in {sleep_for}s"48                )49            sleep(sleep_for)50