Spaces:
Sleeping
Sleeping
File size: 2,947 Bytes
2415446 | 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 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 | """Shared API request validation and safe error logging."""
from typing import Any, Literal
from fastapi import HTTPException
from fastapi.responses import JSONResponse
from loguru import logger
from free_claude_code.application.errors import ApplicationError, InvalidRequestError
from free_claude_code.config.settings import Settings
from free_claude_code.core.anthropic import (
anthropic_error_payload,
anthropic_error_type_for_failure,
)
from free_claude_code.core.diagnostics import (
redacted_exception_traceback,
safe_exception_message,
)
from free_claude_code.core.openai_responses import (
openai_error_payload,
openai_error_type_for_failure,
)
WireApi = Literal["messages", "responses"]
def require_non_empty_messages(messages: list[Any]) -> None:
if not messages:
raise InvalidRequestError("messages cannot be empty")
def ordinary_application_error_response(
error: ApplicationError,
*,
wire_api: WireApi,
request_id: str,
) -> JSONResponse:
"""Serialize a deterministic application error without terminal headers."""
if wire_api == "responses":
return JSONResponse(
status_code=error.status_code,
content=openai_error_payload(
message=error.message,
error_type=openai_error_type_for_failure(error.kind),
),
)
return JSONResponse(
status_code=error.status_code,
content=anthropic_error_payload(
error_type=anthropic_error_type_for_failure(error.kind),
message=error.message,
request_id=request_id,
),
)
def http_status_for_unexpected_api_exception(_exc: BaseException) -> int:
return 500
def log_unexpected_api_exception(
settings: Settings,
exc: BaseException,
*,
context: str,
request_id: str | None = None,
) -> None:
"""Log API failures without echoing exception text unless opted in."""
if settings.log_api_error_tracebacks:
if request_id is not None:
logger.error(
"{} request_id={}: {}",
context,
request_id,
safe_exception_message(exc),
)
else:
logger.error("{}: {}", context, safe_exception_message(exc))
logger.error(redacted_exception_traceback(exc))
return
if request_id is not None:
logger.error(
"{} request_id={} exc_type={}",
context,
request_id,
type(exc).__name__,
)
else:
logger.error("{} exc_type={}", context, type(exc).__name__)
def unexpected_http_exception(
settings: Settings, exc: Exception, *, context: str
) -> HTTPException:
log_unexpected_api_exception(settings, exc, context=context)
return HTTPException(
status_code=http_status_for_unexpected_api_exception(exc),
detail=safe_exception_message(exc),
)
|