mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
51 lines
1.7 KiB
Python
51 lines
1.7 KiB
Python
import time
|
|
from http import HTTPStatus
|
|
from logging import Logger
|
|
from typing import Callable, Dict, Optional, TypeVar, cast
|
|
|
|
from utils.logger import setup_logger
|
|
|
|
T = TypeVar("T")
|
|
|
|
RETRYABLE_STATUSES = {
|
|
HTTPStatus.TOO_MANY_REQUESTS,
|
|
HTTPStatus.INTERNAL_SERVER_ERROR,
|
|
HTTPStatus.BAD_GATEWAY,
|
|
HTTPStatus.SERVICE_UNAVAILABLE,
|
|
HTTPStatus.GATEWAY_TIMEOUT,
|
|
}
|
|
|
|
|
|
def call_with_retry(
|
|
fn: Callable[[], T],
|
|
max_retries: int = 2,
|
|
logger: Optional[Logger] = None,
|
|
) -> T:
|
|
"""
|
|
Call fn(), retrying up to max_retries times on 429 rate-limit errors
|
|
or transient 5xx server errors.
|
|
Waits the minimal amount: the remaining interval window reported by HubSpot headers.
|
|
Falls back to the full interval (10s) if headers are absent.
|
|
|
|
Note: each HubSpot sub-module (deals, companies, etc.) ships its own ApiException
|
|
class with no shared base beyond Exception, so we detect retryable statuses via duck-typing.
|
|
"""
|
|
logger = logger or setup_logger()
|
|
for attempt in range(max_retries + 1):
|
|
try:
|
|
return fn()
|
|
except Exception as e:
|
|
status = getattr(e, "status", None)
|
|
if status not in RETRYABLE_STATUSES or attempt == max_retries:
|
|
raise
|
|
headers = cast(Dict[str, str], getattr(e, "headers", None) or {})
|
|
interval_ms = int(
|
|
headers.get("x-hubspot-ratelimit-interval-milliseconds", 10000)
|
|
)
|
|
wait_s = interval_ms / 1000.0
|
|
logger.warning(
|
|
f"HubSpot {status} (attempt {attempt + 1}/{max_retries}), "
|
|
f"waiting {wait_s:.1f}s before retry."
|
|
)
|
|
time.sleep(wait_s)
|
|
raise RuntimeError("Unreachable") # pragma: no cover
|