Spaces:
Running
Running
| """User-facing error formatting shared by API, providers, and integrations.""" | |
| import httpx | |
| import openai | |
| def get_user_facing_error_message( | |
| e: Exception, | |
| *, | |
| read_timeout_s: float | None = None, | |
| ) -> str: | |
| """Return a readable, non-empty error message for users. | |
| Known transport and OpenAI SDK exception types are mapped to stable wording | |
| before falling back to ``str(e)``, so empty or noisy SDK messages do not skip | |
| the mapped path. | |
| """ | |
| if isinstance(e, httpx.ReadTimeout): | |
| if read_timeout_s is not None: | |
| return f"Provider request timed out after {read_timeout_s:g}s." | |
| return "Provider request timed out." | |
| if isinstance(e, httpx.ConnectTimeout): | |
| return "Could not connect to provider." | |
| if isinstance(e, TimeoutError): | |
| if read_timeout_s is not None: | |
| return f"Provider request timed out after {read_timeout_s:g}s." | |
| return "Request timed out." | |
| if isinstance(e, openai.RateLimitError): | |
| return "Provider rate limit reached. Please retry shortly." | |
| if isinstance(e, openai.AuthenticationError): | |
| raw = str(e) | |
| return f"Provider authentication failed. Check API key. Details: {raw[:100]}" | |
| if isinstance(e, openai.BadRequestError): | |
| raw = str(e) | |
| return f"Invalid request sent to provider. Details: {raw[:100]}" | |
| name = type(e).__name__ | |
| status_code = getattr(e, "status_code", None) | |
| if name == "RateLimitError": | |
| return "Provider rate limit reached. Please retry shortly." | |
| if name == "AuthenticationError": | |
| return "Provider authentication failed. Check API key." | |
| if name == "InvalidRequestError": | |
| return "Invalid request sent to provider." | |
| if name == "OverloadedError": | |
| return "Provider is currently overloaded. Please retry." | |
| if name == "APIError": | |
| raw = str(e) | |
| if status_code in (502, 503, 504): | |
| return "Provider is temporarily unavailable. Please retry." | |
| return f"Provider API request failed. Details: {raw[:150]}" | |
| if name.endswith("ProviderError") or name == "ProviderError": | |
| return "Provider request failed." | |
| message = str(e).strip() | |
| if message: | |
| return message | |
| return "Provider request failed unexpectedly." | |
| def format_user_error_preview(exc: Exception, *, max_len: int = 200) -> str: | |
| """Truncate a user-facing error string for short chat replies.""" | |
| return get_user_facing_error_message(exc)[:max_len] | |
| def append_request_id(message: str, request_id: str | None) -> str: | |
| """Append request_id suffix when available.""" | |
| base = message.strip() or "Provider request failed unexpectedly." | |
| if request_id: | |
| return f"{base} (request_id={request_id})" | |
| return base | |