Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/bug_report_config.py: 96%

1from __future__ import annotations 

2 

3import ast 

4import functools 

5import inspect 

6from collections.abc import Mapping, Sequence 

7from pathlib import Path 

8from types import MappingProxyType 

9from typing import Final 

10 

11from pydantic import JsonValue, TypeAdapter, ValidationError 

12 

13import litellm 

14from litellm.litellm_core_utils.bug_report import ( 

15 KNOWN_PROVIDERS, 

16 BugReport, 

17 EnvironmentReport, 

18 allowlisted, 

19 build_bug_report, 

20 build_environment_report, 

21) 

22from litellm.proxy._types import ConfigGeneralSettings 

23from litellm.router_utils.routing_groups import VALID_ROUTING_STRATEGIES 

24from litellm.types.caching import LiteLLMCacheType 

25from litellm.types.guardrails import GuardrailEventHooks, LitellmParams, SupportedGuardrailIntegrations 

26from litellm.types.secret_managers.main import KeyManagementSystem 

27 

28_OBJECT_MAP: Final[TypeAdapter[Mapping[str, object]]] = TypeAdapter(Mapping[str, object]) 

29_OBJECT_LIST: Final[TypeAdapter[tuple[object, ...]]] = TypeAdapter(tuple[object, ...]) 

30_JSON: Final[TypeAdapter[JsonValue]] = TypeAdapter(JsonValue) 

31CREDENTIAL_KEY_PARTS: Final = frozenset( 

32 { 

33 "key", 

34 "keys", 

35 "secret", 

36 "secrets", 

37 "token", 

38 "password", 

39 "passwd", 

40 "credential", 

41 "credentials", 

42 "url", 

43 "uri", 

44 "dsn", 

45 "host", 

46 "hosts", 

47 "base", 

48 "endpoint", 

49 "cert", 

50 "pem", 

51 "salt", 

52 } 

53) 

54ENUM_KEYS_WITH_CREDENTIAL_PARTS: Final = frozenset({"key_management_system"}) 

55 

56 

57def _object_map(value: object) -> Mapping[str, object]: 

58 try: 

59 return _OBJECT_MAP.validate_python(value) 

60 except ValidationError: 

61 return MappingProxyType({}) 

62 

63 

64def _object_list(value: object) -> Sequence[object]: 

65 try: 

66 return _OBJECT_LIST.validate_python(value) 

67 except ValidationError: 

68 return () 

69 

70 

71@functools.cache 

72def _known_values() -> frozenset[str]: 

73 from litellm.litellm_core_utils.custom_logger_registry import CustomLoggerRegistry 

74 

75 return frozenset( 

76 ( 

77 *VALID_ROUTING_STRATEGIES, 

78 *KNOWN_PROVIDERS, 

79 *litellm._known_custom_logger_compatible_callbacks, # pyright: ignore[reportPrivateUsage, reportUnknownMemberType, reportUnknownArgumentType] # untyped List of the callback Literal's args, no public alias 

80 *CustomLoggerRegistry.CALLBACK_CLASS_STR_TO_CLASS_TYPE, 

81 *(member.value for member in LiteLLMCacheType), 

82 *(member.value for member in KeyManagementSystem), 

83 *(member.value for member in SupportedGuardrailIntegrations), 

84 *(member.value for member in GuardrailEventHooks), 

85 ) 

86 ) 

87 

88 

89def _module_level_names(node: ast.stmt) -> tuple[str, ...]: 

90 match node: 

91 case ast.Assign(targets=targets): 

92 return tuple(target.id for target in targets if isinstance(target, ast.Name)) 

93 case ast.AnnAssign(target=ast.Name(id=name)): 

94 return (name,) 

95 case ast.ImportFrom(names=aliases): 

96 return tuple(alias.asname or alias.name for alias in aliases) 

97 case _: 

98 return () 

99 

100 

101@functools.cache 

102def _litellm_settings_keys() -> frozenset[str]: 

103 tree: Final = ast.parse(Path(litellm.__file__).read_text()) 

104 return frozenset(name for node in tree.body for name in _module_level_names(node)) 

105 

106 

107@functools.cache 

108def _router_settings_keys() -> frozenset[str]: 

109 from litellm.router import Router 

110 

111 return frozenset(name for name in inspect.signature(Router.__init__).parameters if name != "self") # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # untyped params, only names are read 

112 

113 

114@functools.cache 

115def _cache_params_keys() -> frozenset[str]: 

116 from litellm.caching.caching import Cache 

117 

118 return frozenset(name for name in inspect.signature(Cache.__init__).parameters if name != "self") # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # untyped params, only names are read 

119 

120 

121def _is_credential_key(key: str) -> bool: 

122 return key not in ENUM_KEYS_WITH_CREDENTIAL_PARTS and not CREDENTIAL_KEY_PARTS.isdisjoint(key.lower().split("_")) 

123 

124 

125def _render_json(key: str, value: JsonValue) -> str | None: 

126 match value: 

127 case bool(): 

128 return str(value).lower() 

129 case str(): 

130 return value if value in _known_values() and not _is_credential_key(key) else None 

131 case list(): 

132 known_items: Final = tuple(rendered for item in value if (rendered := _render_json(key, item)) is not None) 

133 return f"[{', '.join(known_items)}]" if known_items else None 

134 case _: 

135 return None 

136 

137 

138def _render(key: str, value: object) -> str | None: 

139 try: 

140 return _render_json(key, _JSON.validate_python(value)) 

141 except ValidationError: 

142 return None 

143 

144 

145def _section_lines(section: str, values: Mapping[str, object], known_keys: frozenset[str]) -> tuple[str, ...]: 

146 return tuple( 

147 f"{section}.{key} = {rendered}" 

148 for key, value in values.items() 

149 if key in known_keys and (rendered := _render(key, value)) is not None 

150 ) 

151 

152 

153def _guardrail_lines(guardrails: object) -> tuple[str, ...]: 

154 known_keys: Final = frozenset(LitellmParams.model_fields) 

155 return tuple( 

156 line 

157 for index, guardrail in enumerate(_object_list(guardrails)) 

158 for line in _section_lines( 

159 f"guardrails[{index}].litellm_params", _object_map(_object_map(guardrail).get("litellm_params")), known_keys 

160 ) 

161 ) 

162 

163 

164def _deployment_provider(model: object) -> str | None: 

165 prefix: Final = model.split("/", 1)[0] if isinstance(model, str) and "/" in model else None 

166 return allowlisted(prefix, KNOWN_PROVIDERS) 

167 

168 

169def _model_list_lines(model_list: object) -> tuple[str, ...]: 

170 providers: Final = tuple( 

171 sorted( 

172 frozenset( 

173 provider 

174 for deployment in _object_list(model_list) 

175 if ( 

176 provider := _deployment_provider( 

177 _object_map(_object_map(deployment).get("litellm_params")).get("model") 

178 ) 

179 ) 

180 is not None 

181 ) 

182 ) 

183 ) 

184 return (f"model_list[*].provider = [{', '.join(providers)}]",) if providers else () 

185 

186 

187def safe_config_lines(config: Mapping[str, object], general_settings: Mapping[str, object]) -> tuple[str, ...]: 

188 litellm_settings: Final = _object_map(config.get("litellm_settings")) 

189 return ( 

190 *_section_lines("general_settings", general_settings, frozenset(ConfigGeneralSettings.model_fields)), 

191 *_section_lines("litellm_settings", litellm_settings, _litellm_settings_keys()), 

192 *_section_lines( 

193 "litellm_settings.cache_params", _object_map(litellm_settings.get("cache_params")), _cache_params_keys() 

194 ), 

195 *_section_lines("router_settings", _object_map(config.get("router_settings")), _router_settings_keys()), 

196 *_guardrail_lines(config.get("guardrails")), 

197 *_model_list_lines(config.get("model_list")), 

198 ) 

199 

200 

201def _proxy_config_lines() -> tuple[str, ...]: 

202 from litellm.proxy import proxy_server 

203 

204 return safe_config_lines( 

205 proxy_server.proxy_config.config, 

206 _object_map(proxy_server.general_settings), # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # bare dict global, validated by _object_map 

207 ) 

208 

209 

210def build_proxy_environment_report() -> EnvironmentReport: 

211 return build_environment_report(surface="proxy", config_lines=_proxy_config_lines()) 

212 

213 

214def build_proxy_bug_report( 

215 exc: BaseException, 

216 *, 

217 call_type: str | None = None, 

218 custom_llm_provider: object = None, 

219 stream: object = None, 

220) -> BugReport: 

221 return build_bug_report( 

222 exc, 

223 surface="proxy", 

224 call_type=call_type, 

225 custom_llm_provider=custom_llm_provider, 

226 stream=stream, 

227 config_lines=_proxy_config_lines(), 

228 )