Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/list_api/common.py: 92%

1"""Contract machinery shared by every LiteLLM-defined list route, on any surface.""" 

2 

3from collections.abc import Sequence 

4from typing import Final 

5from urllib.parse import urlencode 

6 

7from fastapi import Request 

8from fastapi.dependencies.utils import get_flat_params 

9from fastapi.params import ParamTypes 

10from fastapi.responses import JSONResponse 

11from typing_extensions import ReadOnly, TypedDict 

12 

13from litellm.types.proxy.management_endpoints.management_v1 import ( 

14 ListLinks, 

15 PageLinks, 

16 ProblemDetail, 

17) 

18 

19PROBLEM_CONTENT_TYPE: Final = "application/problem+json" 

20# A URN, not an https URL: RFC 9457 only asks that `type` identify the problem 

21# type, and an https URI promises documentation at that address. Switch to an 

22# https base only when pages actually exist to serve. 

23PROBLEM_TYPE_BASE: Final = "urn:litellm:error:" 

24 

25 

26class ManagementProblem(Exception): 

27 """Raised to return an RFC 9457 problem instead of the proxy's OpenAI error shape.""" 

28 

29 def __init__(self, problem: ProblemDetail) -> None: 

30 self.problem = problem 

31 super().__init__(problem.detail) 

32 

33 

34def problem_response(problem: ProblemDetail) -> JSONResponse: 

35 return JSONResponse( 

36 status_code=problem.status, 

37 content=problem.model_dump(exclude_none=True), 

38 media_type=PROBLEM_CONTENT_TYPE, 

39 ) 

40 

41 

42def _declared_query_params(request: Request) -> frozenset[str]: 

43 route: Final = request.scope.get("route") 

44 dependant: Final = getattr(route, "dependant", None) 

45 if dependant is None: 45 ↛ 46line 45 didn't jump to line 46 because the condition on line 45 was never true

46 return frozenset() 

47 # fastapi>=0.140.7 removed get_flat_dependant(); get_flat_params() returns the 

48 # flattened (deduped) param list. Filter to query params to match the old behavior. 

49 return frozenset( 

50 field.alias 

51 for field in get_flat_params(dependant) 

52 if getattr(field.field_info, "in_", None) == ParamTypes.query 

53 ) 

54 

55 

56def escape_like(value: str) -> str: 

57 """Escape LIKE/ILIKE metacharacters. Ids routinely contain `_`, which is a wildcard unescaped.""" 

58 return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") 

59 

60 

61class ValidationErrorDetail(TypedDict): 

62 """The keys of a pydantic/FastAPI validation error a problem document needs.""" 

63 

64 type: ReadOnly[str] 

65 loc: ReadOnly[tuple[int | str, ...]] 

66 msg: ReadOnly[str] 

67 

68 

69def _is_length_error_of_rejected_items(error: ValidationErrorDetail, errors: Sequence[ValidationErrorDetail]) -> bool: 

70 """pydantic counts only items that validated, so a bad item also trips the parent's min_length.""" 

71 return error["type"] == "too_short" and any( 

72 len(other["loc"]) > len(error["loc"]) and other["loc"][: len(error["loc"])] == error["loc"] for other in errors 

73 ) 

74 

75 

76def request_validation_problem(raw_errors: Sequence[ValidationErrorDetail]) -> ProblemDetail: 

77 """A body that fails validation (an unknown field included) is 422; a bad query parameter is 400.""" 

78 errors: Final = tuple(error for error in raw_errors if not _is_length_error_of_rejected_items(error, raw_errors)) 

79 detail: Final = "; ".join(f"{'.'.join(str(part) for part in error['loc'][1:])}: {error['msg']}" for error in errors) 

80 if any(error["loc"] and error["loc"][0] == "body" for error in errors): 

81 return ProblemDetail( 

82 type=f"{PROBLEM_TYPE_BASE}invalid-request-body", 

83 title="Invalid request body", 

84 status=422, 

85 detail=detail or "The request body is invalid.", 

86 ) 

87 return ProblemDetail( 

88 type=f"{PROBLEM_TYPE_BASE}invalid-query-parameter", 

89 title="Invalid query parameter", 

90 status=400, 

91 detail=detail or "The request query parameters are invalid.", 

92 ) 

93 

94 

95def unknown_query_param_problem(unknown: tuple[str, ...], allowed: tuple[str, ...]) -> ProblemDetail: 

96 return ProblemDetail( 

97 type=f"{PROBLEM_TYPE_BASE}unknown-query-parameter", 

98 title="Unknown query parameter", 

99 status=400, 

100 detail=f"Unrecognized query parameter(s): {', '.join(unknown)}.", 

101 allowed=sorted(allowed), 

102 ) 

103 

104 

105async def reject_unknown_query_params(request: Request) -> None: 

106 """Reject any query param the route did not declare. 

107 

108 A silently ignored filter over-returns data, which is worse than a rejected 

109 request; a fresh surface is the only chance to be strict about it. 

110 """ 

111 declared: Final = _declared_query_params(request) 

112 unknown: Final[tuple[str, ...]] = tuple(sorted(name for name in request.query_params if name not in declared)) 

113 if not unknown: 113 ↛ 115line 113 didn't jump to line 115 because the condition on line 113 was always true

114 return 

115 raise ManagementProblem(unknown_query_param_problem(unknown=unknown, allowed=tuple(sorted(declared)))) 

116 

117 

118def _page_url(request: Request, page: int) -> str: 

119 others: Final = tuple((key, value) for key, value in request.query_params.multi_items() if key != "page") 

120 return f"{request.url.path}?{urlencode((*others, ('page', page)))}" 

121 

122 

123def build_page_links(request: Request, page: int, has_more: bool) -> PageLinks: 

124 return PageLinks( 

125 self_link=_page_url(request, page), 

126 prev=_page_url(request, page - 1) if page > 1 else None, 

127 next=_page_url(request, page + 1) if has_more else None, 

128 ) 

129 

130 

131def build_list_links(request: Request, page: int, total_pages: int) -> ListLinks: 

132 """Page-mode links. `last` clamps to page 1 on an empty result set so every link still resolves.""" 

133 last: Final = max(total_pages, 1) 

134 return ListLinks( 

135 self_link=_page_url(request, page), 

136 first=_page_url(request, 1), 

137 prev=_page_url(request, page - 1) if page > 1 else None, 

138 next=_page_url(request, page + 1) if page < last else None, 

139 last=_page_url(request, last), 

140 )