Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/config_resolvers/settings_store.py: 78%

1from __future__ import annotations 

2 

3from collections.abc import Iterator, Mapping, MutableMapping 

4from types import MappingProxyType 

5from typing import Final 

6 

7from litellm.proxy.config_resolvers._descriptors import FieldSource 

8from litellm.proxy.config_resolvers.settings_rules import ( 

9 ABSENT, 

10 Absent, 

11 DbRow, 

12 JsonValue, 

13 Resolved, 

14 Section, 

15 SettingValue, 

16 is_resource_list, 

17 resolve, 

18 rule_for, 

19) 

20 

21 

22class ConfigOwnedKeyError(RuntimeError): 

23 def __init__(self, section: Section, key: str, *, shadows_db_value: bool = False) -> None: 

24 super().__init__(config_ownership_message(section=section, key=key, shadows_db_value=shadows_db_value)) 

25 self.section: Final = section 

26 self.key: Final = key 

27 self.shadows_db_value: Final = shadows_db_value 

28 

29 

30def config_ownership_message(*, section: Section, key: str, shadows_db_value: bool) -> str: 

31 stored: Final = ( 

32 " The value stored in the database for it is ignored and will never be applied." if shadows_db_value else "" 

33 ) 

34 return ( 

35 f"{section}.{key} is set in the config file, so the config file owns it and it cannot be changed " 

36 f"here.{stored} Edit the config file to change it, or remove it from the file to let the database own it." 

37 ) 

38 

39 

40_EMPTY_VALUES: Final[Mapping[str, JsonValue]] = MappingProxyType({}) 

41_EMPTY_ROWS: Final[Mapping[DbRow, Mapping[str, JsonValue]]] = MappingProxyType({}) 

42 

43 

44class SettingsStore(MutableMapping[str, JsonValue]): 

45 def __init__(self, section: Section) -> None: 

46 self._section: Final = section 

47 self._yaml_values: Mapping[str, JsonValue] = _EMPTY_VALUES 

48 self._database_rows: Mapping[DbRow, Mapping[str, JsonValue]] = _EMPTY_ROWS 

49 self._runtime_values: Mapping[str, JsonValue] = _EMPTY_VALUES 

50 self._deleted_runtime_keys: frozenset[str] = frozenset() 

51 

52 def load_yaml(self, mapping: Mapping[str, JsonValue]) -> None: 

53 self._yaml_values = MappingProxyType( 

54 {key: value for key, value in mapping.items() if not is_resource_list(self._section, key)} 

55 ) 

56 self._runtime_values = MappingProxyType( 

57 {key: value for key, value in self._runtime_values.items() if is_resource_list(self._section, key)} 

58 ) 

59 self._deleted_runtime_keys = frozenset() 

60 

61 def config_value(self, key: str) -> JsonValue: 

62 return self._yaml_values.get(key) 

63 

64 def owned_by_config(self, key: str) -> bool: 

65 return key in self._yaml_values 

66 

67 def rejected_writes(self, incoming: Mapping[str, JsonValue]) -> tuple[str, ...]: 

68 return tuple( 

69 sorted(key for key, value in incoming.items() if self.owned_by_config(key) and value != self.get(key)) 

70 ) 

71 

72 def shadowed_db_keys(self) -> tuple[str, ...]: 

73 """Keys the config file owns whose stored value differs, so the stored one never reaches a reader.""" 

74 return tuple(sorted(key for key in self._yaml_values if self._db_value_is_shadowed(key))) 

75 

76 def shadows_db_value(self, key: str) -> bool: 

77 return self.owned_by_config(key) and self._db_value_is_shadowed(key) 

78 

79 def apply_db_row(self, row: DbRow, db_row: Mapping[str, JsonValue]) -> None: 

80 previous_row: Final = self._database_rows.get(row, _EMPTY_VALUES) 

81 changed: Final = frozenset( 

82 key 

83 for key in (*previous_row, *db_row) 

84 if previous_row.get(key, ABSENT) != db_row.get(key, ABSENT) # pyright: ignore[reportUnknownArgumentType] # JsonValue vs Absent compare 

85 ) 

86 self._database_rows = MappingProxyType({**self._database_rows, row: MappingProxyType(dict(db_row))}) 

87 self._clear_runtime_keys(changed) 

88 

89 def resolved(self) -> Mapping[str, JsonValue]: 

90 return MappingProxyType(dict(self)) 

91 

92 def apply_runtime_values(self, values: Mapping[str, JsonValue]) -> None: 

93 self._runtime_values = MappingProxyType(dict(values)) 

94 self._deleted_runtime_keys = frozenset() 

95 

96 def source(self, key: str) -> FieldSource: 

97 return self._resolution_for(key).source 

98 

99 def __getitem__(self, key: str) -> JsonValue: 

100 if key in self._deleted_runtime_keys: 100 ↛ 101line 100 didn't jump to line 101 because the condition on line 100 was never true

101 raise KeyError(key) 

102 if key in self._runtime_values: 

103 return self._runtime_values[key] 

104 resolved: Final = self._resolution_for(key) 

105 if isinstance(resolved.value, Absent): 

106 raise KeyError(key) 

107 return resolved.value 

108 

109 def __setitem__(self, key: str, value: JsonValue) -> None: 

110 if self.owned_by_config(key) and value != self.get(key): 110 ↛ 111line 110 didn't jump to line 111 because the condition on line 110 was never true

111 raise ConfigOwnedKeyError(self._section, key, shadows_db_value=self._db_value_is_shadowed(key)) 

112 self._runtime_values = MappingProxyType({**self._runtime_values, key: value}) 

113 self._deleted_runtime_keys = self._deleted_runtime_keys - frozenset((key,)) 

114 

115 def __delitem__(self, key: str) -> None: 

116 if key not in self: 

117 raise KeyError(key) 

118 if self.owned_by_config(key): 

119 raise ConfigOwnedKeyError(self._section, key, shadows_db_value=self._db_value_is_shadowed(key)) 

120 self._runtime_values = MappingProxyType( 

121 {key_: value for key_, value in self._runtime_values.items() if key_ != key} 

122 ) 

123 self._deleted_runtime_keys = self._deleted_runtime_keys | frozenset((key,)) 

124 

125 def clear(self) -> None: 

126 self._deleted_runtime_keys = frozenset(key for key in self._keys() if not self.owned_by_config(key)) 

127 self._runtime_values = MappingProxyType( 

128 {key: value for key, value in self._runtime_values.items() if self.owned_by_config(key)} 

129 ) 

130 

131 def __iter__(self) -> Iterator[str]: 

132 return iter( 

133 key 

134 for key in self._keys() 

135 if key not in self._deleted_runtime_keys 

136 and (key in self._runtime_values or not isinstance(self._resolution_for(key).value, Absent)) 

137 ) 

138 

139 def __len__(self) -> int: 

140 return sum(1 for _ in self) 

141 

142 def __bool__(self) -> bool: 

143 return any(True for _ in self) 

144 

145 def _clear_runtime_keys(self, keys: frozenset[str]) -> None: 

146 stale: Final = frozenset(key for key in keys if not self.owned_by_config(key)) 

147 if not stale: 

148 return 

149 self._runtime_values = MappingProxyType( 

150 {key: value for key, value in self._runtime_values.items() if key not in stale} 

151 ) 

152 self._deleted_runtime_keys = self._deleted_runtime_keys - stale 

153 

154 def _keys(self) -> tuple[str, ...]: 

155 return tuple( 

156 dict.fromkeys( 

157 ( 

158 *self._yaml_values, 

159 *(key for row in self._database_rows.values() for key in row), 

160 *self._runtime_values, 

161 ) 

162 ) 

163 ) 

164 

165 def db_value(self, key: str) -> SettingValue: 

166 return self._db_value(key) if is_resource_list(self._section, key) else ABSENT 

167 

168 def _db_value(self, key: str) -> SettingValue: 

169 rule: Final = rule_for(self._section, key) 

170 return self._database_rows.get(rule.db_row, _EMPTY_VALUES).get(key, ABSENT) 

171 

172 def _db_value_is_shadowed(self, key: str) -> bool: 

173 db_value: Final = self._db_value(key) 

174 return ( 

175 not isinstance(db_value, Absent) 

176 and db_value is not None 

177 and db_value != self.get(key) 

178 and db_value != self.config_value(key) 

179 ) 

180 

181 def _resolution_for(self, key: str) -> Resolved: 

182 yaml_value: Final[SettingValue] = self._yaml_values.get(key, ABSENT) 

183 return resolve(yaml_value, self._db_value(key)) 

184 

185 

186def source_for(settings: SettingsStore, key: str, default: object = None) -> FieldSource: 

187 source: Final = settings.source(key) 

188 if source == "unset": 188 ↛ 190line 188 didn't jump to line 190 because the condition on line 188 was always true

189 return "default" if default is not None else "unset" 

190 return source