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

1"""Shapes the ``error`` object the proxy answers with so it matches OpenAI's contract: 

2``type`` is a required string and ``param`` is nullable, neither of which the literal 

3string ``"None"`` satisfies.""" 

4 

5from collections.abc import Mapping 

6from types import MappingProxyType 

7from typing import Final 

8 

9from fastapi import status 

10 

11from litellm.constants import STRINGIFIED_NONE 

12from litellm.proxy._types import ProxyException 

13 

14LITELLM_CALL_ID_HEADER: Final = "x-litellm-call-id" 

15 

16_OPENAI_ERROR_TYPE_BY_STATUS: Final[Mapping[int, str]] = MappingProxyType( 

17 { 

18 status.HTTP_401_UNAUTHORIZED: "authentication_error", 

19 status.HTTP_403_FORBIDDEN: "permission_error", 

20 status.HTTP_429_TOO_MANY_REQUESTS: "rate_limit_error", 

21 } 

22) 

23 

24 

25def attribute_of(value: object, name: str, default: object = None) -> object: 

26 return getattr(value, name, default) 

27 

28 

29def error_status_code(exc: object, default: int) -> int: 

30 """The HTTP status an exception carries as ``status_code`` or, the way ``ProxyException`` 

31 stores it, as a stringified ``code``; ``default`` when it carries neither.""" 

32 carried: Final = attribute_of(exc, "status_code") 

33 if isinstance(carried, int) and not isinstance(carried, bool): 

34 return carried 

35 stringified: Final = attribute_of(exc, "code") 

36 return int(stringified) if isinstance(stringified, str) and stringified.isdecimal() else default 

37 

38 

39def openai_error_type(exc: object, status_code: int) -> str: 

40 """OpenAI types ``error.type`` as a required string, so an exception carrying none 

41 falls back to the type its status code stands for.""" 

42 carried: Final = attribute_of(exc, "type") 

43 if isinstance(carried, str) and carried != STRINGIFIED_NONE: 

44 return carried 

45 mapped: Final = _OPENAI_ERROR_TYPE_BY_STATUS.get(status_code) 

46 if mapped is not None: 

47 return mapped 

48 if status_code < status.HTTP_500_INTERNAL_SERVER_ERROR: 

49 return "invalid_request_error" 

50 return "internal_server_error" 

51 

52 

53def openai_error_param(exc: object) -> str | None: 

54 """OpenAI types ``error.param`` as nullable, so an exception carrying none 

55 serializes as JSON ``null``.""" 

56 carried: Final = attribute_of(exc, "param") 

57 return carried if isinstance(carried, str) and carried != STRINGIFIED_NONE else None 

58 

59 

60def litellm_call_id_headers(litellm_call_id: str | None) -> dict[str, str] | None: # mutable-ok: ProxyException.headers 

61 if litellm_call_id is None: 

62 return None 

63 return {LITELLM_CALL_ID_HEADER: litellm_call_id} # mutable-ok: ProxyException mutates its headers dict 

64 

65 

66def with_litellm_call_id(exc: ProxyException, litellm_call_id: str | None) -> ProxyException: 

67 """The same error object, answering with ``x-litellm-call-id`` when it was raised without one.""" 

68 if litellm_call_id is not None: 

69 exc.headers.setdefault(LITELLM_CALL_ID_HEADER, litellm_call_id) 

70 return exc 

71 

72 

73def headers_with_litellm_call_id(headers: Mapping[str, str] | None, litellm_call_id: str) -> Mapping[str, str]: 

74 """``headers`` plus ``x-litellm-call-id``, keeping the value they already carry under that name.""" 

75 if headers is None: 

76 return MappingProxyType({LITELLM_CALL_ID_HEADER: litellm_call_id}) 

77 return MappingProxyType({LITELLM_CALL_ID_HEADER: litellm_call_id, **headers})