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

1import os 

2from collections.abc import Awaitable, Mapping 

3from types import MappingProxyType 

4from typing import Final, Protocol 

5 

6from litellm._logging import verbose_proxy_logger 

7 

8INCLUDE_KEY: Final = "include" 

9 

10 

11def resolve_include_file_path(include_file: str, declared_in: str, root_config_path: str) -> str: 

12 """ 

13 Resolve one `include` entry to the file it names, next to the config that declares it. 

14 

15 A config written before nested entries resolved this way can name a file sitting next to the root 

16 config instead, so that file is still read, with a warning naming where it was found. When both 

17 files exist the one next to the declaring config wins and the other is named in a warning. 

18 """ 

19 declared_relative: Final = os.path.abspath(os.path.join(os.path.dirname(declared_in), include_file)) 

20 root_relative: Final = os.path.abspath(os.path.join(os.path.dirname(root_config_path), include_file)) 

21 if root_relative == declared_relative or not os.path.exists(root_relative): 

22 return declared_relative 

23 

24 if not os.path.exists(declared_relative): 

25 verbose_proxy_logger.warning( 

26 "Config include '%s' declared in %s was not found next to it, so %s was read instead. " 

27 "Move the included file next to the config that declares it.", 

28 include_file, 

29 declared_in, 

30 root_relative, 

31 ) 

32 return root_relative 

33 

34 verbose_proxy_logger.warning( 

35 "Config include '%s' declared in %s matches two files. %s sits next to that config and was read, " 

36 "so %s was skipped. Rename one of the two to say which one you meant.", 

37 include_file, 

38 declared_in, 

39 declared_relative, 

40 root_relative, 

41 ) 

42 return declared_relative 

43 

44 

45class IncludeResolver(Protocol): 

46 def __call__(self, include_entry: str, declared_in: str, /) -> str: ... 46 ↛ exitline 46 didn't return from function '__call__' because

47 

48 

49class ConfigReader(Protocol): 

50 def __call__(self, location: str, /) -> Awaitable[Mapping[str, object]]: ... 50 ↛ exitline 50 didn't return from function '__call__' because

51 

52 

53def _merged_value(base_value: object, included_value: object) -> object: 

54 if isinstance(included_value, list) and isinstance(base_value, list): 

55 return [*base_value, *included_value] # mutable-ok: a merged config value stays the plain list the proxy loads 

56 return included_value 

57 

58 

59def _merged_entry(base: Mapping[str, object], included: Mapping[str, object], key: str) -> object: 

60 if key not in included: 

61 return base[key] 

62 return _merged_value(base.get(key), included[key]) 

63 

64 

65def _merged(base: Mapping[str, object], included: Mapping[str, object]) -> Mapping[str, object]: 

66 return MappingProxyType({key: _merged_entry(base, included, key) for key in (*base, *included)}) 

67 

68 

69def _without_include(config: Mapping[str, object]) -> Mapping[str, object]: 

70 return MappingProxyType({key: value for key, value in config.items() if key != INCLUDE_KEY}) 

71 

72 

73def include_entries(config: Mapping[str, object]) -> tuple[str, ...]: 

74 if INCLUDE_KEY not in config: 74 ↛ 77line 74 didn't jump to line 77 because the condition on line 74 was always true

75 return () 

76 

77 entries: Final = config[INCLUDE_KEY] 

78 if not isinstance(entries, list): 

79 raise ValueError("'include' must be a list of file paths") 

80 

81 paths: Final = tuple(entry for entry in entries if isinstance(entry, str)) 

82 if len(paths) != len(entries): 

83 raise ValueError("'include' must be a list of file paths") 

84 

85 return paths 

86 

87 

88def _pending_from(config: Mapping[str, object], location: str) -> tuple[tuple[str, str], ...]: 

89 return tuple((entry, location) for entry in include_entries(config)) 

90 

91 

92async def _resolve( 

93 config: Mapping[str, object], 

94 pending: tuple[tuple[str, str], ...], 

95 loaded: frozenset[str], 

96 resolve: IncludeResolver, 

97 read: ConfigReader, 

98) -> Mapping[str, object]: 

99 if not pending: 99 ↛ 102line 99 didn't jump to line 102 because the condition on line 99 was always true

100 return _without_include(config) 

101 

102 entry, declared_in = pending[0] 

103 location: Final = resolve(entry, declared_in) 

104 if location in loaded: 

105 return await _resolve(config, pending[1:], loaded, resolve, read) 

106 

107 included: Final = await read(location) 

108 return await _resolve( 

109 _merged(config, _without_include(included)), 

110 (*pending[1:], *_pending_from(included, location)), 

111 loaded | frozenset((location,)), 

112 resolve, 

113 read, 

114 ) 

115 

116 

117async def resolve_includes( 

118 config: Mapping[str, object], 

119 location: str, 

120 resolve: IncludeResolver, 

121 read: ConfigReader, 

122) -> dict[str, object]: 

123 """ 

124 Merge every config named by the `include` directive into the config that declares it. 

125 

126 List values are extended and every other value is overridden, `resolve` turns each entry into the 

127 location it names relative to the config that declares it, a config already pulled in is neither 

128 read nor merged a second time, and `read` decides where a location is read from, so the same merge 

129 applies to configs on disk and to configs hosted in a bucket. 

130 """ 

131 merged: Final = await _resolve(config, _pending_from(config, location), frozenset((location,)), resolve, read) 

132 return dict(merged) # mutable-ok: the proxy mutates the config it loads