mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-19 17:03:02 +00:00
45 lines
1.6 KiB
Python
45 lines
1.6 KiB
Python
import json
|
|
from typing import Any, Callable, Dict, Optional, TypeVar, cast
|
|
|
|
from infrastructure.hubspot.errors import HubspotRequestError
|
|
from infrastructure.hubspot.retry import call_with_retry
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
def scrubbed(error: Exception) -> HubspotRequestError:
|
|
"""Reduce an SDK error to non-PII diagnostics, dropping the body.
|
|
|
|
Each HubSpot sub-module ships its own ApiException class with no shared
|
|
base beyond Exception, so the fields are read by duck-typing.
|
|
"""
|
|
status = getattr(error, "status", None)
|
|
status_code = status if isinstance(status, int) else None
|
|
|
|
category: Optional[str] = None
|
|
correlation_id: Optional[str] = None
|
|
try:
|
|
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
|
|
|
|
return HubspotRequestError(
|
|
status_code=status_code, category=category, correlation_id=correlation_id
|
|
)
|
|
|
|
|
|
def scrubbed_call_with_retry(fn: Callable[[], T]) -> T:
|
|
"""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
|