Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/common_utils/responses_stream_errors.py: 36%

1import time 

2from collections.abc import Mapping 

3from http import HTTPStatus 

4from types import MappingProxyType 

5from typing import Final 

6 

7from pydantic import BaseModel, ConfigDict, field_validator 

8 

9from litellm._logging import redact_internal_details_from_client_message 

10from litellm._uuid import uuid 

11from litellm.exceptions import MidStreamFallbackError 

12from litellm.types.llms.openai import ResponseFailedEvent, ResponsesAPIResponse, ResponsesAPIStreamEvents 

13 

14 

15class _ResponseIdentity(BaseModel): 

16 model_config = ConfigDict(frozen=True, from_attributes=True) 

17 

18 id: str | None = None 

19 model: str | None = None 

20 created_at: int | None = None 

21 

22 

23class _StreamEvent(BaseModel): 

24 model_config = ConfigDict(frozen=True, from_attributes=True) 

25 

26 type: str | None = None 

27 sequence_number: int | None = None 

28 response: _ResponseIdentity | None = None 

29 

30 

31class _FailureDetails(BaseModel): 

32 model_config = ConfigDict(frozen=True, from_attributes=True) 

33 

34 message: str | None = None 

35 code: str | int | None = None 

36 type: str | None = None 

37 status_code: int | None = None 

38 

39 @field_validator("message", mode="before") 

40 @classmethod 

41 def normalize_message(cls, value: object) -> str | None: 

42 return value if isinstance(value, str) else None 

43 

44 @field_validator("code", mode="before") 

45 @classmethod 

46 def normalize_code(cls, value: object) -> str | int | None: 

47 return value if isinstance(value, (str, int)) and not isinstance(value, bool) else None 

48 

49 @field_validator("type", mode="before") 

50 @classmethod 

51 def normalize_type(cls, value: object) -> str | None: 

52 return value if isinstance(value, str) else None 

53 

54 

55def _original_failure(exception: Exception) -> Exception: 

56 current = exception # rebind-ok: the recursion gate requires iterative wrapper traversal 

57 while isinstance(current, MidStreamFallbackError) and current.original_exception is not None: 

58 current = current.original_exception 

59 return current 

60 

61 

62def _failure_details(original: Exception) -> _FailureDetails: 

63 mapped: Final = _FailureDetails.model_validate(original) 

64 body: Final = getattr(original, "body", None) 

65 if not isinstance(body, Mapping): 

66 return mapped 

67 upstream: Final = _FailureDetails.model_validate(body) 

68 return _FailureDetails( 

69 message=upstream.message or mapped.message, 

70 code=upstream.code if upstream.code is not None else mapped.code, 

71 type=upstream.type or mapped.type, 

72 status_code=mapped.status_code, 

73 ) 

74 

75 

76_CLIENT_ERROR_CODES: Final = MappingProxyType( 

77 { 

78 int(HTTPStatus.UNAUTHORIZED): "authentication_error", 

79 int(HTTPStatus.FORBIDDEN): "permission_error", 

80 int(HTTPStatus.NOT_FOUND): "not_found_error", 

81 int(HTTPStatus.REQUEST_TIMEOUT): "request_timeout", 

82 int(HTTPStatus.TOO_MANY_REQUESTS): "rate_limit_exceeded", 

83 } 

84) 

85 

86 

87def _status_error_code(status_code: int | None) -> str: 

88 if status_code is None or not HTTPStatus.BAD_REQUEST <= status_code < HTTPStatus.INTERNAL_SERVER_ERROR: 

89 return "server_error" 

90 return _CLIENT_ERROR_CODES.get(status_code, "invalid_request_error") 

91 

92 

93def _response_error_code(details: _FailureDetails) -> str: 

94 for value in (details.code, details.type): 

95 if value == "insufficient_quota": 

96 return "insufficient_quota" 

97 if value in (429, "429") or isinstance(value, str) and value.startswith("rate_limit"): 

98 return "rate_limit_exceeded" 

99 if isinstance(details.code, str) and details.code and not details.code.isdecimal(): 

100 return details.code 

101 return _status_error_code(details.status_code) 

102 

103 

104class ResponsesStreamErrorState: 

105 def __init__(self) -> None: 

106 self.response_id: str | None = None 

107 self.model: str | None = None 

108 self.created_at: int | None = None 

109 self.sequence_number = -1 

110 self.terminal_emitted = False 

111 self._pending_event: _StreamEvent | None = None 

112 

113 def observe_chunk(self, chunk: object) -> None: 

114 self._pending_event = _StreamEvent.model_validate(chunk) if isinstance(chunk, (BaseModel, Mapping)) else None 

115 

116 def mark_emitted(self, frame: str | bytes) -> str | bytes: 

117 event: Final = self._pending_event 

118 if event is None: 

119 return frame 

120 if event.sequence_number is not None: 

121 self.sequence_number = max(self.sequence_number, event.sequence_number) 

122 if event.response is not None: 

123 self.response_id = event.response.id or self.response_id 

124 self.model = event.response.model or self.model 

125 if event.response.created_at is not None: 

126 self.created_at = event.response.created_at 

127 if event.type in ("response.completed", "response.failed", "response.incomplete"): 

128 self.terminal_emitted = True 

129 return frame 

130 

131 def format_failure(self, exception: Exception) -> str | None: 

132 if self.terminal_emitted: 

133 return None 

134 original: Final = _original_failure(exception) 

135 details: Final = _failure_details(original) 

136 response: Final = ResponsesAPIResponse.model_validate( 

137 MappingProxyType( 

138 { 

139 "id": self.response_id or f"resp_{uuid.uuid4().hex}", 

140 "object": "response", 

141 "created_at": self.created_at if self.created_at is not None else int(time.time()), 

142 "model": self.model, 

143 "status": "failed", 

144 "output": (), 

145 "error": MappingProxyType( 

146 { 

147 "code": _response_error_code(details), 

148 "message": redact_internal_details_from_client_message(details.message or str(original)), 

149 } 

150 ), 

151 } 

152 ) 

153 ) 

154 event: Final = ResponseFailedEvent.model_validate( 

155 MappingProxyType( 

156 { 

157 "type": ResponsesAPIStreamEvents.RESPONSE_FAILED, 

158 "response": response, 

159 "sequence_number": self.sequence_number + 1, 

160 } 

161 ) 

162 ) 

163 payload: Final = event.model_dump_json(exclude_none=True) 

164 self.terminal_emitted = True 

165 return f"event: response.failed\ndata: {payload}\n\n"