zai / app /core /httpx_errors.py
bg5's picture
Upload 85 files
e0b68d2 verified
Raw History Blame Contribute Delete
2.14 kB
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Helpers for normalized httpx error logging."""
from __future__ import annotations
from typing import TYPE_CHECKING, Optional
if TYPE_CHECKING:
from app.core.openai_compat import get_error_message
from app.core.retry_policy import summarize_upstream_error_text
_DEFAULT_MAX_BODY = 240
def _format_context(context: str) -> str:
if not context:
return ""
return f"[{context}] "
def _sanitize_text(text: str, max_length: int = _DEFAULT_MAX_BODY) -> str:
if not text:
return ""
sanitized = " ".join(text.split())
if len(sanitized) > max_length:
return sanitized[:max_length].rstrip() + "..."
return sanitized
def _format_attempt(attempt: Optional[int]) -> str:
if attempt is None:
return ""
return f" attempt={attempt}"
def normalize_httpx_exception(
exc: Exception,
*,
method: str,
url: str,
context: str = "",
attempt: Optional[int] = None,
) -> str:
"""Normalize httpx exception into a short log message."""
from app.core.openai_compat import get_error_message
friendly = get_error_message(exc)
if not friendly:
friendly = repr(exc)
return (
f"{_format_context(context)}HTTP {method.upper()} {url}"
f" failed{_format_attempt(attempt)}: {friendly}"
)
def normalize_httpx_response(
status_code: int,
body_text: str,
*,
content_type: Optional[str] = None,
method: str,
url: str,
context: str = "",
) -> str:
"""Normalize non-200 response into a short log message."""
from app.core.retry_policy import summarize_upstream_error_text
summarized = summarize_upstream_error_text(
status_code,
body_text or "",
content_type=content_type,
)
summarized = _sanitize_text(summarized, max_length=_DEFAULT_MAX_BODY)
message = f"{_format_context(context)}HTTP {method.upper()} {url} returned {status_code}"
if summarized:
return f"{message}: {summarized}"
return message