From 8d9fb0ade4915a0cd43dffa5bf4ddeb093d94f39 Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Mon, 6 Jul 2026 13:38:20 +0000 Subject: [PATCH] =?UTF-8?q?Transient=20HubSpot=20failures=20are=20retried?= =?UTF-8?q?=20under=20the=20shared=20rate-limit=20policy=20=F0=9F=9F=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5 --- etl/hubspot/deal_contacts_client.py | 33 +++++++++++-------- etl/hubspot/hubspotClient.py | 39 ++-------------------- etl/hubspot/retry.py | 51 +++++++++++++++++++++++++++++ 3 files changed, 73 insertions(+), 50 deletions(-) create mode 100644 etl/hubspot/retry.py diff --git a/etl/hubspot/deal_contacts_client.py b/etl/hubspot/deal_contacts_client.py index fe2516b84..c218b6e05 100644 --- a/etl/hubspot/deal_contacts_client.py +++ b/etl/hubspot/deal_contacts_client.py @@ -6,6 +6,7 @@ from hubspot.crm.associations.v4 import AssociationSpec # type: ignore[reportMi from hubspot.crm.contacts import SimplePublicObjectInputForCreate # type: ignore[reportMissingTypeStubs] from etl.hubspot.errors import HubspotRequestError +from etl.hubspot.retry import call_with_retry # HubSpot-defined deal -> contact association ("deal_to_contact" in v3 terms). DEAL_TO_CONTACT_ASSOCIATION_TYPE_ID = 3 @@ -49,9 +50,11 @@ class HubspotDealContactsClient: def create_contact(self, properties: Dict[str, str]) -> str: try: - contact = self._client.crm.contacts.basic_api.create( # type: ignore[reportUnknownMemberType] - simple_public_object_input_for_create=SimplePublicObjectInputForCreate( - properties=properties + contact = call_with_retry( + lambda: self._client.crm.contacts.basic_api.create( # type: ignore[reportUnknownMemberType] + simple_public_object_input_for_create=SimplePublicObjectInputForCreate( + properties=properties + ) ) ) except Exception as error: @@ -61,17 +64,19 @@ class HubspotDealContactsClient: def associate_contact_to_deal(self, deal_id: str, contact_id: str) -> None: try: - self._client.crm.associations.v4.basic_api.create( # type: ignore[reportUnknownMemberType] - object_type="deals", - object_id=deal_id, - to_object_type="contacts", - to_object_id=contact_id, - association_spec=[ - AssociationSpec( - association_category="HUBSPOT_DEFINED", - association_type_id=DEAL_TO_CONTACT_ASSOCIATION_TYPE_ID, - ) - ], + call_with_retry( + lambda: self._client.crm.associations.v4.basic_api.create( # type: ignore[reportUnknownMemberType] + object_type="deals", + object_id=deal_id, + to_object_type="contacts", + to_object_id=contact_id, + association_spec=[ + AssociationSpec( + association_category="HUBSPOT_DEFINED", + association_type_id=DEAL_TO_CONTACT_ASSOCIATION_TYPE_ID, + ) + ], + ) ) except Exception as error: raise _scrubbed(error) from None diff --git a/etl/hubspot/hubspotClient.py b/etl/hubspot/hubspotClient.py index 918701cac..6ee71a381 100644 --- a/etl/hubspot/hubspotClient.py +++ b/etl/hubspot/hubspotClient.py @@ -1,7 +1,5 @@ import os -import time from enum import Enum -from http import HTTPStatus from typing import Optional, cast, Callable, Any from hubspot.client import Client # type: ignore[reportMissingTypeStubs] @@ -32,6 +30,7 @@ from hubspot.crm.associations.v4.models import ( # type: ignore[reportMissingTy from backend.app.config import get_settings from etl.hubspot.company_data import CompanyData from etl.hubspot.project_data import ProjectData +from etl.hubspot.retry import call_with_retry from utils.logger import setup_logger import mimetypes @@ -87,40 +86,8 @@ class HubspotClient: # self.client def _call_with_retry(self, fn: Callable[[], Any], max_retries: int = 2) -> Any: - """ - 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. - """ - retryable_statuses = { - HTTPStatus.TOO_MANY_REQUESTS, - HTTPStatus.INTERNAL_SERVER_ERROR, - HTTPStatus.BAD_GATEWAY, - HTTPStatus.SERVICE_UNAVAILABLE, - HTTPStatus.GATEWAY_TIMEOUT, - } - 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 = getattr(e, "headers", None) or {} - interval_ms = int( - headers.get("x-hubspot-ratelimit-interval-milliseconds", 10000) - ) - wait_s = interval_ms / 1000.0 - self.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 + """Delegates to the shared etl.hubspot.retry policy.""" + return call_with_retry(fn, max_retries=max_retries, logger=self.logger) def get_deal_ids_from_company(self, company_id: str) -> list[str]: associations_api: AssociationsBasicApi = ( # type: ignore[reportUnknownMemberType] diff --git a/etl/hubspot/retry.py b/etl/hubspot/retry.py new file mode 100644 index 000000000..d1500600a --- /dev/null +++ b/etl/hubspot/retry.py @@ -0,0 +1,51 @@ +import time +from http import HTTPStatus +from logging import Logger +from typing import Callable, Optional, TypeVar + +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 = 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