52 lines
1.5 KiB
Python
52 lines
1.5 KiB
Python
|
from enum import StrEnum
|
||
|
|
|
||
|
|
|
||
|
|
class IntercomErrorCode(StrEnum):
|
||
|
|
CONFIG_ERROR = "config_error"
|
||
|
|
AUTH_ERROR = "auth_error"
|
||
|
|
RATE_LIMITED = "rate_limited"
|
||
|
|
NOT_FOUND = "not_found"
|
||
|
|
NETWORK_ERROR = "network_error"
|
||
|
|
CONVERSATION_CLOSED = "conversation_closed"
|
||
|
|
CONVERSATION_SNOOZED = "conversation_snoozed"
|
||
|
|
HANDOFF_DETECTED = "handoff_detected"
|
||
|
|
SEND_FAILED = "send_failed"
|
||
|
|
UNKNOWN = "unknown"
|
||
|
|
|
||
|
|
|
||
|
|
class IntercomError(Exception):
|
||
|
|
def __init__(self, code: IntercomErrorCode, message: str, retryable: bool = False):
|
||
|
|
self.code = code
|
||
|
|
self.retryable = retryable
|
||
|
|
super().__init__(f"[{code.value}] {message}")
|
||
|
|
|
||
|
|
|
||
|
|
_ERROR_RETRY_MAP: dict[str, tuple[bool, float | None]] = {
|
||
|
|
"rate_limited": (True, None),
|
||
|
|
"service_unavailable": (True, 1.0),
|
||
|
|
"internal_server_error": (True, 3.0),
|
||
|
|
"timeout": (True, 0.5),
|
||
|
|
"network_error": (True, 1.0),
|
||
|
|
"not_found": (False, None),
|
||
|
|
"unauthorized": (False, None),
|
||
|
|
"forbidden": (False, None),
|
||
|
|
"bad_request": (False, None),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def classify_intercom_error(status_code: int, error_body: str = "") -> tuple[bool, float | None]:
|
||
|
|
if status_code == 429:
|
||
|
|
return (True, None)
|
||
|
|
if status_code >= 500:
|
||
|
|
return (True, 2.0)
|
||
|
|
if status_code in (401, 403):
|
||
|
|
return (False, None)
|
||
|
|
if status_code == 404:
|
||
|
|
return (False, None)
|
||
|
|
return (False, None)
|
||
|
|
|
||
|
|
|
||
|
|
def is_retryable_status(status_code: int) -> bool:
|
||
|
|
retryable, _ = classify_intercom_error(status_code)
|
||
|
|
return retryable
|