File size: 2,136 Bytes
e0b68d2
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
#!/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