| """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 | |