HubSpot calls share one retry policy behind a single guarded entry point 🟪

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

etl/hubspot/deal_contacts_client.py

@@ -1,5 +1,5 @@
 import json
-from typing import Dict, Optional
+from typing import Any, Callable, Dict, Optional, cast
 
 from hubspot.client import Client  # type: ignore[reportMissingTypeStubs]
 from hubspot.crm.associations.v4 import AssociationSpec  # type: ignore[reportMissingTypeStubs]
@@ -24,10 +24,15 @@ def _scrubbed(error: Exception) -> HubspotRequestError:
     category: Optional[str] = None
     correlation_id: Optional[str] = None
     try:
-        details = json.loads(getattr(error, "body", None) or "")
-        if isinstance(details, dict):
-            category = details.get("category")  # type: ignore[reportUnknownMemberType]
-            correlation_id = details.get("correlationId")  # type: ignore[reportUnknownMemberType]
+        parsed: Any = json.loads(getattr(error, "body", None) or "")
+        if isinstance(parsed, dict):
+            details = cast(Dict[str, Any], parsed)
+            raw_category = details.get("category")
+            raw_correlation_id = details.get("correlationId")
+            category = raw_category if isinstance(raw_category, str) else None
+            correlation_id = (
+                raw_correlation_id if isinstance(raw_correlation_id, str) else None
+            )
     except ValueError:
         pass
 
@@ -49,34 +54,36 @@ class HubspotDealContactsClient:
         self._client: Client = sdk_client
 
     def create_contact(self, properties: Dict[str, str]) -> str:
-        try:
-            contact = call_with_retry(
-                lambda: self._client.crm.contacts.basic_api.create(  # type: ignore[reportUnknownMemberType]
-                    simple_public_object_input_for_create=SimplePublicObjectInputForCreate(
-                        properties=properties
-                    )
+        contact = self._call(
+            lambda: self._client.crm.contacts.basic_api.create(  # type: ignore[reportUnknownMemberType]
+                simple_public_object_input_for_create=SimplePublicObjectInputForCreate(
+                    properties=properties
                 )
             )
-        except Exception as error:
-            # `from None` keeps the body-carrying SDK error out of tracebacks.
-            raise _scrubbed(error) from None
+        )
         return str(contact.id)  # type: ignore[reportUnknownMemberType]
 
     def associate_contact_to_deal(self, deal_id: str, contact_id: str) -> None:
-        try:
-            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,
-                        )
-                    ],
-                )
+        self._call(
+            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,
+                    )
+                ],
             )
+        )
+
+    @staticmethod
+    def _call(fn: Callable[[], Any]) -> Any:
+        """Run one SDK call under the shared retry policy, scrubbed on failure."""
+        try:
+            return call_with_retry(fn)
         except Exception as error:
+            # `from None` keeps the body-carrying SDK error out of tracebacks.
             raise _scrubbed(error) from None

etl/hubspot/retry.py

@@ -1,7 +1,7 @@
 import time
 from http import HTTPStatus
 from logging import Logger
-from typing import Callable, Optional, TypeVar
+from typing import Callable, Dict, Optional, TypeVar, cast
 
 from utils.logger import setup_logger
 
@@ -38,7 +38,7 @@ def call_with_retry(
             status = getattr(e, "status", None)
             if status not in RETRYABLE_STATUSES or attempt == max_retries:
                 raise
-            headers = getattr(e, "headers", None) or {}
+            headers = cast(Dict[str, str], getattr(e, "headers", None) or {})
             interval_ms = int(
                 headers.get("x-hubspot-ratelimit-interval-milliseconds", 10000)
             )