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

1""" 

2Contains utils used by OpenAI compatible endpoints 

3""" 

4 

5from typing import Final 

6 

7from fastapi import Request 

8 

9from litellm.litellm_core_utils.sensitive_data_masker import SensitiveDataMasker 

10from litellm.proxy.common_utils.http_parsing_utils import _read_request_body 

11 

12SENSITIVE_DATA_MASKER: Final = SensitiveDataMasker() 

13 

14 

15def remove_sensitive_info_from_deployment( 

16 deployment_dict: dict, 

17 excluded_keys: set[str] | None = None, 

18) -> dict: 

19 """ 

20 Removes sensitive information from a deployment dictionary. 

21 

22 Args: 

23 deployment_dict (dict): The deployment dictionary to remove sensitive information from. 

24 excluded_keys (Optional[Set[str]]): Set of keys that should not be masked (exact match). 

25 

26 Returns: 

27 dict: The modified deployment dictionary with sensitive information removed. 

28 """ 

29 deployment_dict["litellm_params"].pop("api_key", None) 

30 deployment_dict["litellm_params"].pop("client_secret", None) 

31 deployment_dict["litellm_params"].pop("vertex_credentials", None) 

32 deployment_dict["litellm_params"].pop("vertex_ai_credentials", None) 

33 deployment_dict["litellm_params"].pop("aws_access_key_id", None) 

34 deployment_dict["litellm_params"].pop("aws_secret_access_key", None) 

35 

36 # Rate-limit config fields must never be masked — they are integers, not credentials. 

37 # The field names contain "key" which matches the masker's sensitive pattern, so we 

38 # explicitly exclude them here rather than widening the global non_sensitive_overrides. 

39 _rate_limit_config_keys: Final = { 

40 "default_api_key_tpm_limit", 

41 "default_api_key_rpm_limit", 

42 } 

43 _excluded: Final = (excluded_keys or set()) | _rate_limit_config_keys 

44 

45 deployment_dict["litellm_params"] = SENSITIVE_DATA_MASKER.mask_dict( 

46 deployment_dict["litellm_params"], excluded_keys=_excluded 

47 ) 

48 

49 return deployment_dict 

50 

51 

52async def get_custom_llm_provider_from_request_body(request: Request) -> str | None: 

53 """ 

54 Get the `custom_llm_provider` from the request body 

55 

56 Safely reads the request body 

57 """ 

58 request_body: Final[dict] = await _read_request_body(request=request) or {} 

59 if "custom_llm_provider" in request_body: 

60 return request_body["custom_llm_provider"] 

61 return None 

62 

63 

64def get_custom_llm_provider_from_request_query(request: Request) -> str | None: 

65 """ 

66 Get the `custom_llm_provider` from the request query parameters 

67 

68 Safely reads the request query parameters 

69 """ 

70 if "custom_llm_provider" in request.query_params: 70 ↛ 71line 70 didn't jump to line 71 because the condition on line 70 was never true

71 return request.query_params["custom_llm_provider"] 

72 return None 

73 

74 

75def get_custom_llm_provider_from_request_headers(request: Request) -> str | None: 

76 """ 

77 Get the `custom_llm_provider` from the request header `custom-llm-provider` 

78 """ 

79 if "custom-llm-provider" in request.headers: 79 ↛ 80line 79 didn't jump to line 80 because the condition on line 79 was never true

80 return request.headers["custom-llm-provider"] 

81 return None