File size: 1,272 Bytes
c641d5f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 | """Retry helper shared by external tools."""
from __future__ import annotations
import time
from collections.abc import Callable
from typing import TypeVar
T = TypeVar("T")
def with_retry(
operation: Callable[[], T],
attempts: int = 3,
backoff: float = 1.0,
retry_if: Callable[[Exception], bool] | None = None,
) -> T:
last: Exception | None = None
for attempt in range(max(1, attempts)):
try:
return operation()
except Exception as exc: # the caller chooses an idempotent operation
last = exc
if retry_if is not None and not retry_if(exc):
raise
if attempt + 1 < attempts:
time.sleep(backoff * (2**attempt))
assert last is not None
raise last
class RateLimiter:
"""Thread-safe minimum-interval limiter."""
def __init__(self, requests_per_minute: float):
import threading
self.interval = 60.0 / max(1.0, requests_per_minute)
self._next = 0.0
self._lock = threading.Lock()
def wait(self) -> None:
with self._lock:
delay = self._next - time.monotonic()
if delay > 0:
time.sleep(delay)
self._next = time.monotonic() + self.interval
|