diff --git a/litellm/exceptions.py b/litellm/exceptions.py index fd328fdfed6..ceb484015db 100644 --- a/litellm/exceptions.py +++ b/litellm/exceptions.py @@ -425,9 +425,9 @@ class RateLimitError(openai.RateLimitError): # type: ignore def __init__( self, - message, - llm_provider, - model, + message: str, + llm_provider: str, + model: Optional[str], response: Optional[httpx.Response] = None, litellm_debug_info: Optional[str] = None, max_retries: Optional[int] = None, @@ -435,7 +435,7 @@ class RateLimitError(openai.RateLimitError): # type: ignore category: Union[str, RateLimitErrorCategory] = (RateLimitErrorCategory.VENDOR_RATE_LIMIT), rate_limit_type: Optional[Union[str, RateLimitType]] = None, headers: Optional[Dict[str, str]] = None, - detail: Any = None, + detail: object = None, ): self.status_code = 429 self.message = "litellm.RateLimitError: {}".format(message) @@ -509,9 +509,9 @@ class RateLimitError(openai.RateLimitError): # type: ignore class InsufficientQuotaError(RateLimitError): def __init__( self, - message, - llm_provider, - model, + message: str, + llm_provider: str, + model: Optional[str], response: Optional[httpx.Response] = None, litellm_debug_info: Optional[str] = None, max_retries: Optional[int] = None, @@ -519,7 +519,7 @@ class InsufficientQuotaError(RateLimitError): category: Union[str, RateLimitErrorCategory] = (RateLimitErrorCategory.VENDOR_RATE_LIMIT), rate_limit_type: Optional[Union[str, RateLimitType]] = None, headers: Optional[Dict[str, str]] = None, - detail: Any = None, + detail: object = None, ): super().__init__( message=message, diff --git a/litellm/litellm_core_utils/exception_mapping_utils.py b/litellm/litellm_core_utils/exception_mapping_utils.py index e85f6de6a6c..92a0cba1173 100644 --- a/litellm/litellm_core_utils/exception_mapping_utils.py +++ b/litellm/litellm_core_utils/exception_mapping_utils.py @@ -4,6 +4,7 @@ import traceback from typing import Any, Optional, Protocol, cast import httpx +from pydantic import TypeAdapter import litellm from litellm._logging import _ENABLE_SECRET_REDACTION, _redact_string, verbose_logger @@ -84,9 +85,6 @@ class ExceptionCheckers: Returns: True if the error indicates insufficient quota, False otherwise """ - if not isinstance(error_str, str): - return False - return "insufficient_quota" in error_str.lower() @staticmethod @@ -201,6 +199,9 @@ def _get_body_error_code(error_str: str) -> int | None: return None +_ERROR_BODY_ADAPTER: TypeAdapter[dict[object, object]] = TypeAdapter(dict[object, object]) + + def _body_has_insufficient_quota_code(error_obj: object) -> bool: """True only when the provider's STRUCTURED error body carries code or type == "insufficient_quota". @@ -211,26 +212,25 @@ def _body_has_insufficient_quota_code(error_obj: object) -> bool: missing or unparseable body returns False, so callers degrade to the plain retryable RateLimitError rather than promoting on a free-text match. """ - try: - if error_obj is None: - return False - if getattr(error_obj, "code", None) == "insufficient_quota": - return True - if getattr(error_obj, "type", None) == "insufficient_quota": - return True - body = getattr(error_obj, "body", None) - if not isinstance(body, dict): - return False - nested = body.get("error") - candidates = ( - body.get("code"), - body.get("type"), - nested.get("code") if isinstance(nested, dict) else None, - nested.get("type") if isinstance(nested, dict) else None, - ) - return "insufficient_quota" in candidates - except Exception: + if error_obj is None: return False + if getattr(error_obj, "code", None) == "insufficient_quota": + return True + if getattr(error_obj, "type", None) == "insufficient_quota": + return True + raw_body = getattr(error_obj, "body", None) + if not isinstance(raw_body, dict): + return False + body = _ERROR_BODY_ADAPTER.validate_python(raw_body) + raw_nested = body.get("error") + nested = _ERROR_BODY_ADAPTER.validate_python(raw_nested) if isinstance(raw_nested, dict) else None + candidates = ( + body.get("code"), + body.get("type"), + nested.get("code") if nested is not None else None, + nested.get("type") if nested is not None else None, + ) + return "insufficient_quota" in candidates def _get_response_headers(original_exception: Exception) -> Optional[httpx.Headers]: diff --git a/tests/test_litellm/litellm_core_utils/test_exception_mapping_utils.py b/tests/test_litellm/litellm_core_utils/test_exception_mapping_utils.py index fa5e48f953c..0a58045271c 100644 --- a/tests/test_litellm/litellm_core_utils/test_exception_mapping_utils.py +++ b/tests/test_litellm/litellm_core_utils/test_exception_mapping_utils.py @@ -749,9 +749,6 @@ class TestIsErrorStrInsufficientQuota: is False ) - def test_non_string_returns_false(self): - assert ExceptionCheckers.is_error_str_insufficient_quota(None) is False # type: ignore - def test_openai_insufficient_quota_maps_to_insufficient_quota_error(): """