scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-19 00:01:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 88

refactor(proxy): resolve config and DB settings precedence in one SettingsStore

Browse source
This commit is contained in:
Yuneng Jiang 2026-09-17 23:22:19 -07:00
parent 1be40e304d
commit d1cd869012
No known key found for this signature in database
25 changed files with 17481 additions and 1088 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
litellm/proxy/_experimental/mcp_server/discoverable_endpoints.py
View file

@ -2579,7 +2579,7 @@ def _jwt_auth_issuers() -> list:
if env_issuer:
issuers.append(env_issuer)
jwtauth: Final = general_settings.get("litellm_jwtauth") if isinstance(general_settings, dict) else None
jwtauth: Final = general_settings.get("litellm_jwtauth") if isinstance(general_settings, Mapping) else None
raw_issuers: Final = jwtauth.get("issuers") if isinstance(jwtauth, dict) else getattr(jwtauth, "issuers", None)
for cfg in raw_issuers or []:
issuer = cfg.get("issuer") if isinstance(cfg, dict) else getattr(cfg, "issuer", None)

3
litellm/proxy/client/cli/commands/autoroute/commands.py
View file

@ -2,6 +2,7 @@ import atexit
import secrets
import signal
import threading
from collections.abc import Mapping
from types import FrameType
from typing import Final
@ -67,7 +68,7 @@ def _ensure_master_key() -> str:
master_key: Final = secrets.token_urlsafe(32)
general_settings: Final = generated.get("general_settings")
updated_settings: Final[dict[str, JsonValue]] = {
**(general_settings if isinstance(general_settings, dict) else {}),
**(general_settings if isinstance(general_settings, Mapping) else {}),
"master_key": master_key,
}
updated: Final[dict[str, JsonValue]] = {**generated, "general_settings": updated_settings}

3
litellm/proxy/client/cli/commands/autoroute/config.py
View file

@ -1,3 +1,4 @@
from collections.abc import Mapping
from typing import Final, Literal
from pydantic import BaseModel, ConfigDict, Field, JsonValue, TypeAdapter
@ -221,7 +222,7 @@ def master_key_from_config(config: dict[str, JsonValue]) -> str | None:
normalized copy here would diverge from what the proxy expects.
"""
general_settings: Final = config.get("general_settings")
if not isinstance(general_settings, dict):
if not isinstance(general_settings, Mapping):
return None
master_key: Final = general_settings.get("master_key")
if isinstance(master_key, str) and master_key.strip():

2
litellm/proxy/config_resolvers/__init__.py
View file

@ -7,4 +7,4 @@ from litellm.proxy.config_resolvers._descriptors import (
)
from litellm.proxy.config_resolvers.settings_store import SettingsStore
__all__ = ["FieldDescriptor", "FieldSource", "SettingsStore", "resolve_fields"]
__all__ = ("FieldDescriptor", "FieldSource", "SettingsStore", "resolve_fields")

4
litellm/proxy/config_resolvers/_descriptors.py
View file

@ -69,5 +69,7 @@ def resolve_fields(
"""
resolved: Final = tuple(_resolve_one(descriptor, db_values, env, empty_db_is_set) for descriptor in descriptors)
values: Final = {field_name: value for field_name, value, _ in resolved}
provenance: Final[dict[str, FieldSource]] = dict((field_name, source) for field_name, _, source in resolved)
provenance: Final[dict[str, FieldSource]] = dict( # mutable-ok: public resolver contract returns a plain dict
(field_name, source) for field_name, _, source in resolved
)
return values, provenance

88
litellm/proxy/config_resolvers/settings_rules.py
View file

@ -94,29 +94,35 @@ _UI_SETTINGS_FIELDS: Final[tuple[str, ...]] = (
)
DUAL_SOURCE_KEYS: Final[Mapping[tuple[Section, str], KeyRule]] = MappingProxyType(
{
**{
("general_settings", key): KeyRule(db_row="general_settings", kind="db_wins")
for key in _DB_GENERAL_SETTINGS
},
**{
("general_settings", key): KeyRule(db_row="general_settings", kind="config_wins")
for key in _CONFIG_GENERAL_SETTINGS
},
**{
("general_settings", key): KeyRule(db_row="general_settings", kind="db_fallback_to_config")
for key in _CLEANUP_BOUNDS
},
("general_settings", "alerting"): KeyRule(db_row="general_settings", kind="list_union"),
("general_settings", "pass_through_endpoints"): KeyRule(db_row="general_settings", kind="merge_by_path"),
("general_settings", "*"): KeyRule(db_row="general_settings", kind="db_overlay"),
("router_settings", "*"): KeyRule(db_row="router_settings", kind="db_overlay"),
("litellm_settings", "*"): KeyRule(db_row="litellm_settings", kind="db_overlay"),
("environment_variables", "*"): KeyRule(db_row="environment_variables", kind="db_overlay"),
**{("general_settings", key): KeyRule(db_row="ui_settings", kind="db_wins") for key in _UI_SETTINGS_FIELDS},
}
)
def _rules_for(
section: Section, keys: tuple[str, ...], db_row: DbRow, kind: RuleKind
) -> tuple[tuple[tuple[Section, str], KeyRule], ...]:
return tuple(((section, key), KeyRule(db_row=db_row, kind=kind)) for key in keys)
def _build_dual_source_keys() -> Mapping[tuple[Section, str], KeyRule]:
return MappingProxyType(
dict(
(
*_rules_for("general_settings", _DB_GENERAL_SETTINGS, "general_settings", "db_wins"),
*_rules_for("general_settings", _CONFIG_GENERAL_SETTINGS, "general_settings", "config_wins"),
*_rules_for("general_settings", _CLEANUP_BOUNDS, "general_settings", "db_fallback_to_config"),
(("general_settings", "alerting"), KeyRule(db_row="general_settings", kind="list_union")),
(
("general_settings", "pass_through_endpoints"),
KeyRule(db_row="general_settings", kind="merge_by_path"),
),
(("general_settings", "*"), KeyRule(db_row="general_settings", kind="db_overlay")),
(("router_settings", "*"), KeyRule(db_row="router_settings", kind="db_overlay")),
(("litellm_settings", "*"), KeyRule(db_row="litellm_settings", kind="db_overlay")),
(("environment_variables", "*"), KeyRule(db_row="environment_variables", kind="db_overlay")),
*_rules_for("general_settings", _UI_SETTINGS_FIELDS, "ui_settings", "db_wins"),
)
)
)
DUAL_SOURCE_KEYS: Final[Mapping[tuple[Section, str], KeyRule]] = _build_dual_source_keys()
def rule_for(section: Section, key: str) -> KeyRule:
@ -168,7 +174,11 @@ def _list_union(yaml_value: SettingValue, db_value: SettingValue) -> Resolved:
return _db_wins(yaml_value, db_value)
if not isinstance(yaml_value, list) or not isinstance(db_value, list):
return _db_wins(yaml_value, db_value)
return Resolved(value=[*yaml_value, *(value for value in db_value if value not in yaml_value)], source="db")
merged: Final[list[JsonValue]] = [ # mutable-ok: resolved config values retain the legacy JSON-list contract
*yaml_value,
*(value for value in db_value if value not in yaml_value),
]
return Resolved(value=merged, source="db")
def _merge_by_path(yaml_value: SettingValue, db_value: SettingValue) -> Resolved:
@ -177,14 +187,18 @@ def _merge_by_path(yaml_value: SettingValue, db_value: SettingValue) -> Resolved
if not isinstance(yaml_value, list) or not isinstance(db_value, list):
return _db_wins(yaml_value, db_value)
db_paths: Final = frozenset(_endpoint_path(value) for value in db_value if _endpoint_path(value) is not None)
return Resolved(
value=[*db_value, *(value for value in yaml_value if _endpoint_path(value) not in db_paths)], source="db"
)
merged: Final[list[JsonValue]] = [ # mutable-ok: resolved config values retain the legacy JSON-list contract
*db_value,
*(value for value in yaml_value if _endpoint_path(value) not in db_paths),
]
return Resolved(value=merged, source="db")
def _db_overlay(yaml_value: SettingValue, db_value: SettingValue) -> Resolved:
if not _db_is_present(db_value):
return _db_wins(yaml_value, db_value)
if isinstance(db_value, list) and not db_value and yaml_value is not ABSENT:
return Resolved(value=yaml_value, source="config")
if not isinstance(yaml_value, dict) or not isinstance(db_value, dict):
return _db_wins(yaml_value, db_value)
overlay: Final = _overlay_mapping(yaml_value, db_value)
@ -193,13 +207,19 @@ def _db_overlay(yaml_value: SettingValue, db_value: SettingValue) -> Resolved:
def _overlay_mapping(yaml_value: dict[str, JsonValue], db_value: dict[str, JsonValue]) -> dict[str, JsonValue]:
retained: Final = {
key: _overlay_value(value, db_value[key]) if key in db_value else value for key, value in yaml_value.items()
}
additions: Final = {
key: value for key, value in db_value.items() if key not in yaml_value and not _db_overlay_defers(value)
}
return {**retained, **additions}
return dict( # mutable-ok: resolved config values retain the legacy JSON-object contract
(
*(
(key, _overlay_value(value, db_value[key]) if key in db_value else value)
for key, value in yaml_value.items()
),
*(
(key, value)
for key, value in db_value.items()
if key not in yaml_value and not _db_overlay_defers(value)
),
)
)
def _overlay_value(yaml_value: JsonValue, db_value: JsonValue) -> JsonValue:

26
litellm/proxy/config_resolvers/settings_store.py
View file

@ -34,12 +34,17 @@ class SettingsStore(MutableMapping[str, JsonValue]):
self._clear_runtime()
def apply_db_row(self, row: DbRow, db_row: Mapping[str, JsonValue]) -> None:
previous_row: Final = self._database_rows.get(row, _EMPTY_VALUES)
self._database_rows = MappingProxyType({**self._database_rows, row: MappingProxyType(dict(db_row))})
self._clear_runtime()
self._clear_runtime_keys(frozenset((*previous_row, *db_row)))
def resolved(self) -> Mapping[str, JsonValue]:
return MappingProxyType(dict(self))
def apply_runtime_values(self, values: Mapping[str, JsonValue]) -> None:
self._runtime_values = MappingProxyType(dict(values))
self._deleted_runtime_keys = frozenset()
def source(self, key: str) -> FieldSource:
return self._resolution_for(key).source
@ -55,7 +60,7 @@ class SettingsStore(MutableMapping[str, JsonValue]):
def __setitem__(self, key: str, value: JsonValue) -> None:
self._runtime_values = MappingProxyType({**self._runtime_values, key: value})
self._deleted_runtime_keys = self._deleted_runtime_keys - {key}
self._deleted_runtime_keys = self._deleted_runtime_keys - frozenset((key,))
def __delitem__(self, key: str) -> None:
if key not in self:
@ -63,10 +68,15 @@ class SettingsStore(MutableMapping[str, JsonValue]):
self._runtime_values = MappingProxyType(
{key_: value for key_, value in self._runtime_values.items() if key_ != key}
)
self._deleted_runtime_keys = self._deleted_runtime_keys | {key}
self._deleted_runtime_keys = self._deleted_runtime_keys | frozenset((key,))
def __iter__(self) -> Iterator[str]:
return iter(key for key in self._keys() if key not in self._deleted_runtime_keys)
return iter(
key
for key in self._keys()
if key not in self._deleted_runtime_keys
and (key in self._runtime_values or not isinstance(self._resolution_for(key).value, Absent))
)
def __len__(self) -> int:
return sum(1 for _ in self)
@ -75,6 +85,14 @@ class SettingsStore(MutableMapping[str, JsonValue]):
self._runtime_values = _EMPTY_VALUES
self._deleted_runtime_keys = frozenset()
def _clear_runtime_keys(self, keys: frozenset[str]) -> None:
if not keys:
return
self._runtime_values = MappingProxyType(
{key: value for key, value in self._runtime_values.items() if key not in keys}
)
self._deleted_runtime_keys = self._deleted_runtime_keys - keys
def _keys(self) -> tuple[str, ...]:
return tuple(
dict.fromkeys(

2
litellm/proxy/management_endpoints/coordination_redis_endpoints.py
View file

@ -198,7 +198,7 @@ async def _current_coordination_redis_settings() -> dict[str, object] | None:
config_state: Final = _SETTINGS_ADAPTER.validate_python(proxy_config.get_config_state())
general_settings: Final = config_state.get(_GENERAL_SETTINGS_PARAM_NAME)
if not isinstance(general_settings, dict):
if not isinstance(general_settings, Mapping):
return None
from_file: Final = general_settings.get(_COORDINATION_REDIS_KEY)
if isinstance(from_file, dict):

2
litellm/proxy/plugin_routes.py
View file

@ -67,7 +67,7 @@ def _configured_key_header_names() -> frozenset[str]:
except Exception:
return frozenset()
general_settings: Final = getattr(proxy_server, "general_settings", None)
if not isinstance(general_settings, dict):
if not isinstance(general_settings, Mapping):
return frozenset()
name: Final[object] = general_settings.get("litellm_key_header_name")
return frozenset({name.lower()}) if isinstance(name, str) and name else frozenset()

647
litellm/proxy/proxy_server.py
View file

@ -431,20 +431,24 @@ from litellm.proxy.common_utils.user_api_key_cache import (
model_access_group_spend_counter_key,
tag_cache_key,
)
from litellm.proxy.config_resolvers import resolve_fields
from litellm.proxy.config_resolvers import SettingsStore, resolve_fields
from litellm.proxy.config_resolvers.alerting import (
EMAIL_DESCRIPTORS,
MS_TEAMS_DESCRIPTORS,
SLACK_DESCRIPTORS,
)
from litellm.proxy.config_resolvers.changed_section_keys import changed_section_keys
from litellm.proxy.config_resolvers.settings_rules import (
JsonValue as SettingsJsonValue,
)
from litellm.proxy.config_resolvers.settings_rules import (
Section,
coerce_bool,
)
from litellm.proxy.container_endpoints.endpoints import router as container_router
from litellm.proxy.credential_endpoints.endpoints import router as credential_router
from litellm.proxy.db.db_transaction_queue.pod_lock_manager import PodLockManager
from litellm.proxy.db.db_transaction_queue.spend_log_cleanup import (
SPEND_LOG_CLEANUP_BOUND_SETTINGS,
SpendLogCleanup,
)
from litellm.proxy.db.db_transaction_queue.spend_log_cleanup import SpendLogCleanup
from litellm.proxy.db.db_transaction_queue.window_spend_update_queue import (
build_window_spend_transaction,
)
@ -4322,7 +4326,7 @@ def _scrub_guardrail_inner(inner: dict[str, JsonValue]) -> None:
inner["guardrail"] = None
def _scrub_db_overlay_remote_module_loads(section: str, db_value: JsonValue) -> JsonValue:
def _scrub_db_overlay_remote_module_loads(section: str, db_value: object) -> object:
"""Strip ``s3://`` / ``gcs://`` entries from the DB-overlay value for
fields whose contents reach ``get_instance_fn``. The same scheme is
allowed from a YAML config (the documented operator flow) but a
@ -4802,6 +4806,21 @@ class _ConfigWithBaseline(dict[str, object]):
self._baseline = MappingProxyType({key: copy.deepcopy(value) for key, value in config.items()})
_EMPTY_SETTINGS_MAPPING: Final[Mapping[str, SettingsJsonValue]] = MappingProxyType({})
_SETTINGS_MAPPING: Final = TypeAdapter(dict[str, SettingsJsonValue])
def _as_settings_mapping(value: object) -> Mapping[str, SettingsJsonValue]:
if not isinstance(value, Mapping):
return _EMPTY_SETTINGS_MAPPING
return _SETTINGS_MAPPING.validate_python(value)
def _bind_general_settings_store(settings: SettingsStore) -> None:
global general_settings
general_settings = settings # pyright: ignore[reportAssignmentType] # legacy global accepts mappings
class ProxyConfig:
"""
Abstraction class on top of config loading/updating logic. Gives us one place to control all config updating logic.
@ -4825,11 +4844,38 @@ class ProxyConfig:
# whether an existing request predates the prices it just fetched, and re-serving one
# costs a single fetch where skipping one leaves it priced wrong indefinitely
self.model_cost_map_applied_revision: int = 0
# Keys explicitly set in the YAML config file. Used to give YAML
# precedence over stale DB-cached values for these specific keys
# during periodic config reloads (_update_general_settings).
self._yaml_general_settings_keys: set[str] = set() # mutable-ok: populated once at startup, read-only thereafter # fmt: skip
self._yaml_spend_log_cleanup_bounds: dict[str, object] = {} # mutable-ok: snapshot of YAML bounds at load time # fmt: skip
self.settings: Final[SettingsStore] = SettingsStore("general_settings")
self.router_settings: Final[SettingsStore] = SettingsStore("router_settings")
self.litellm_settings: Final[SettingsStore] = SettingsStore("litellm_settings")
self.environment_variables: Final[SettingsStore] = SettingsStore("environment_variables")
self._settings_stores: Final[Mapping[Section, SettingsStore]] = MappingProxyType(
{
"general_settings": self.settings,
"router_settings": self.router_settings,
"litellm_settings": self.litellm_settings,
"environment_variables": self.environment_variables,
}
)
def _load_yaml_settings_stores(self, config: Mapping[str, object]) -> None:
for section, store in self._settings_stores.items():
store.load_yaml(_as_settings_mapping(config.get(section)))
store.apply_db_row(section, _EMPTY_SETTINGS_MAPPING)
def _config_with_resolved_settings(self, config: Mapping[str, object]) -> dict[str, object]:
return { # mutable-ok: get_config preserves the mutable mapping contract used by existing loaders
**config,
**{
section: dict(store.resolved())
for section, store in self._settings_stores.items()
if isinstance(config.get(section), Mapping) or len(store) > 0
},
}
def _apply_resolved_runtime_settings(self, config: Mapping[str, object]) -> None:
for section, store in self._settings_stores.items():
if isinstance(config.get(section), Mapping):
store.apply_runtime_values(_as_settings_mapping(config[section]))
def is_yaml(self, config_file_path: str) -> bool:
if not os.path.isfile(config_file_path):
@ -5372,6 +5418,8 @@ class ProxyConfig:
config = await self._get_config_from_file(config_file_path=config_file_path)
self._load_yaml_settings_stores(config)
## UPDATE CONFIG WITH DB
if prisma_client is not None and store_model_in_db is True:
config = await self._update_config_from_db(
@ -5380,6 +5428,8 @@ class ProxyConfig:
store_model_in_db=store_model_in_db,
)
config = self._config_with_resolved_settings(config)
## PRINT YAML FOR CONFIRMING IT WORKS
printed_yaml: Final = copy.deepcopy(config)
printed_yaml.pop("environment_variables", None)
@ -5387,6 +5437,7 @@ class ProxyConfig:
self._initialize_secret_manager_from_raw_config(config=config, config_file_path=config_file_path)
config = self._check_for_os_environ_vars(config=config)
self._apply_resolved_runtime_settings(config)
self.update_config_state(config=config)
@ -5965,17 +6016,6 @@ class ProxyConfig:
_hc_staleness = None
_hc_ignore_transient = False
if general_settings:
# Record which keys were explicitly set in the YAML config file.
# These keys take precedence over DB-cached values during periodic
# reloads (see _update_general_settings).
self._yaml_general_settings_keys = set(general_settings.keys()) # mutable-ok: snapshot of YAML keys at load time # fmt: skip
# The VALUES matter for the cleanup bounds, not just which keys were
# set: clearing one from the dashboard has to fall back to what the
# YAML declared, and a set of names cannot answer that.
self._yaml_spend_log_cleanup_bounds = { # mutable-ok: snapshot of YAML bounds at load time # fmt: skip
key: general_settings[key] for key in SPEND_LOG_CLEANUP_BOUND_SETTINGS if key in general_settings
}
### LOAD KEY MANAGEMENT SETTINGS ###
# The secret manager itself is brought up by get_config(), which runs before the
# `os.environ/` references in this config were resolved. Re-reading the settings here
@ -6355,7 +6395,8 @@ class ProxyConfig:
## NON-LLM CONFIGS eg. MCP tools, vector stores, etc.
await self._init_non_llm_configs(config=config, config_file_path=config_file_path)
return router, router.get_model_list(), general_settings
_bind_general_settings_store(self.settings)
return router, router.get_model_list(), self.settings
async def _init_non_llm_configs(self, config: dict, config_file_path: str | None = None):
"""
@ -6803,13 +6844,6 @@ class ProxyConfig:
config_data=config_data, llm_router=llm_router, prisma_client=prisma_client
)
# general settings
self._add_general_settings_from_db_config(
config_data=config_data,
general_settings=general_settings,
proxy_logging_obj=proxy_logging_obj,
)
return still_desired_ids
def _add_callback_from_db_to_in_memory_litellm_callbacks(
@ -7016,121 +7050,25 @@ class ProxyConfig:
async def _add_router_settings_from_db_config(
self,
config_data: dict,
config_data: Mapping[str, object],
llm_router: Router | None,
prisma_client: PrismaClient | None,
) -> None:
"""
Adds router settings from DB config to litellm proxy
1. Get router settings from DB
2. Get router settings from config
3. Combine both
4. Update router settings
"""
if llm_router is not None and prisma_client is not None:
db_router_settings: Final[_ConfigParamRow | None] = await _config_param_table(prisma_client).find_first(
where={"param_name": "router_settings"}
)
config_router_settings: Final = config_data.get("router_settings", {})
combined_router_settings = {}
if (
config_router_settings is not None
and isinstance(config_router_settings, dict)
and db_router_settings is not None
and isinstance(db_router_settings.param_value, dict)
):
from litellm.utils import _update_dictionary
db_overlay_deferring_empty_lists_to_config: Final = {
k: v
for k, v in db_router_settings.param_value.items()
if not (k in config_router_settings and isinstance(v, list) and len(v) == 0)
}
combined_router_settings = _update_dictionary(
config_router_settings, db_overlay_deferring_empty_lists_to_config
)
elif config_router_settings is not None and isinstance(config_router_settings, dict):
combined_router_settings = config_router_settings
elif db_router_settings is not None and isinstance(db_router_settings.param_value, dict):
combined_router_settings = db_router_settings.param_value
if combined_router_settings:
llm_router.update_settings(**combined_router_settings)
def _add_general_settings_from_db_config(
self, config_data: dict, general_settings: dict, proxy_logging_obj: ProxyLogging
) -> None:
"""
Adds general settings from DB config to litellm proxy
Args:
config_data: dict
general_settings: dict - global general_settings currently in use
proxy_logging_obj: ProxyLogging
"""
_general_settings: Final = config_data.get("general_settings", {})
if _general_settings is not None and "alerting" in _general_settings:
if (
general_settings is not None
and general_settings.get("alerting", None) is not None
and isinstance(general_settings["alerting"], list)
and _general_settings.get("alerting", None) is not None
and isinstance(_general_settings["alerting"], list)
):
# Merge DB and YAML/config alerting values instead of overriding
_yaml_alerting: Final = set(general_settings["alerting"])
_db_alerting: Final = set(_general_settings["alerting"])
_merged_alerting = list(_yaml_alerting.union(_db_alerting))
# Preserve order: YAML values first, then DB values
_merged_alerting = list(general_settings["alerting"]) + [
item for item in _general_settings["alerting"] if item not in general_settings["alerting"]
]
verbose_proxy_logger.debug(
"Merging alerting values: YAML=%s, DB=%s, Merged=%s",
general_settings["alerting"],
_general_settings["alerting"],
_merged_alerting,
)
general_settings["alerting"] = _merged_alerting
# Use update_values to properly set alerting for both slack and email
proxy_logging_obj.update_values(
alerting=general_settings["alerting"],
)
elif general_settings is None:
general_settings = {}
general_settings["alerting"] = _general_settings["alerting"]
# Use update_values to properly set alerting for both slack and email
proxy_logging_obj.update_values(
alerting=general_settings["alerting"],
)
elif isinstance(general_settings, dict):
general_settings["alerting"] = _general_settings["alerting"]
# Use update_values to properly set alerting for both slack and email
proxy_logging_obj.update_values(
alerting=general_settings["alerting"],
)
if _general_settings is not None and "alert_types" in _general_settings:
general_settings["alert_types"] = _general_settings["alert_types"]
proxy_logging_obj.alert_types = general_settings["alert_types"]
proxy_logging_obj.slack_alerting_instance.update_values(
alert_types=general_settings["alert_types"], llm_router=llm_router
)
if _general_settings is not None and "alert_to_webhook_url" in _general_settings:
general_settings["alert_to_webhook_url"] = _general_settings["alert_to_webhook_url"]
proxy_logging_obj.slack_alerting_instance.update_values(
alert_to_webhook_url=general_settings["alert_to_webhook_url"],
llm_router=llm_router,
)
if _general_settings is not None and "plugins" in _general_settings:
general_settings["plugins"] = _general_settings["plugins"]
register_plugins_from_config(general_settings)
if llm_router is None or prisma_client is None:
return
self.router_settings.load_yaml(_as_settings_mapping(config_data.get("router_settings")))
db_router_settings: Final[_ConfigParamRow | None] = await _config_param_table(prisma_client).find_first(
where={"param_name": "router_settings"}
)
db_values: Final = (
_as_settings_mapping(db_router_settings.param_value)
if db_router_settings is not None and db_router_settings.param_value is not None
else _EMPTY_SETTINGS_MAPPING
)
self.router_settings.apply_db_row("router_settings", db_values)
combined_router_settings: Final = self.router_settings.resolved()
if combined_router_settings:
llm_router.update_settings(**combined_router_settings)
async def _reschedule_spend_log_cleanup_job(self):
"""
@ -7206,260 +7144,142 @@ class ProxyConfig:
except ValueError:
verbose_proxy_logger.error("Invalid maximum_spend_logs_retention_interval value")
async def _update_general_settings(self, db_general_settings: Json | None):
"""
Pull from DB, read general settings value
"""
global general_settings, store_model_in_db
async def _update_general_settings(self, db_general_settings: Mapping[str, SettingsJsonValue] | None) -> None:
global general_settings
if db_general_settings is None:
return
_general_settings: Final = dict(db_general_settings)
## MAX PARALLEL REQUESTS ##
if "max_parallel_requests" in _general_settings:
general_settings["max_parallel_requests"] = _general_settings["max_parallel_requests"]
if not isinstance(general_settings, SettingsStore):
self.settings.load_yaml(_as_settings_mapping(general_settings))
cache_size_was_db: Final = self.settings.source("user_api_key_cache_max_size") == "db"
previous_retention_values: Final = self._resolved_retention_values()
self.settings.apply_db_row("general_settings", db_general_settings)
_bind_general_settings_store(self.settings)
await self._apply_general_settings_side_effects(
db_general_settings,
cache_size_was_db,
previous_retention_values,
)
if "global_max_parallel_requests" in _general_settings:
general_settings["global_max_parallel_requests"] = _general_settings["global_max_parallel_requests"]
if "max_batch_file_size_mb" not in self._yaml_general_settings_keys:
general_settings["max_batch_file_size_mb"] = _general_settings.get("max_batch_file_size_mb")
if "max_file_size_mb" not in self._yaml_general_settings_keys:
general_settings["max_file_size_mb"] = _general_settings.get("max_file_size_mb")
if "allowed_file_extensions" not in self._yaml_general_settings_keys:
general_settings["allowed_file_extensions"] = _general_settings.get("allowed_file_extensions")
if "blocked_file_extensions" not in self._yaml_general_settings_keys:
general_settings["blocked_file_extensions"] = _general_settings.get("blocked_file_extensions")
## ALERTING ARGS ##
if "alerting_args" in _general_settings:
general_settings["alerting_args"] = _general_settings["alerting_args"]
proxy_logging_obj.slack_alerting_instance.update_values(
alerting_args=general_settings["alerting_args"],
def _resolved_retention_values(self) -> tuple[SettingsJsonValue | None, ...]:
return tuple(
self.settings.get(key)
for key in (
"maximum_spend_logs_retention_period",
"maximum_autorouter_session_retention_period",
"maximum_health_check_retention_period",
)
)
## PASS-THROUGH ENDPOINTS ##
if "pass_through_endpoints" in _general_settings:
db_pass_through_endpoints: Final = _general_settings["pass_through_endpoints"]
db_pass_through_paths: Final = frozenset(
endpoint.get("path") for endpoint in db_pass_through_endpoints if isinstance(endpoint, dict)
)
general_settings["pass_through_endpoints"] = [
*db_pass_through_endpoints,
*(
endpoint
for endpoint in config_passthrough_endpoints or ()
if endpoint.get("path") not in db_pass_through_paths
),
]
await initialize_pass_through_endpoints(pass_through_endpoints=db_pass_through_endpoints)
## UI ACCESS MODE ##
if "ui_access_mode" in _general_settings:
general_settings["ui_access_mode"] = _general_settings["ui_access_mode"]
## STORE PROMPTS IN SPEND LOGS ##
if "store_prompts_in_spend_logs" in _general_settings:
# If the YAML config explicitly set this key, prefer the YAML value
# over the DB-cached value. This ensures config changes deployed via
# CI/CD take effect without requiring a manual /config/update call.
# When YAML does not set this key, the DB value is used (preserving
# admin UI runtime changes).
if "store_prompts_in_spend_logs" in self._yaml_general_settings_keys:
value = general_settings.get("store_prompts_in_spend_logs")
else:
value = _general_settings["store_prompts_in_spend_logs"]
# Normalize case: handle True/true/TRUE, False/false/FALSE, None/null
if value is None:
general_settings["store_prompts_in_spend_logs"] = None
elif isinstance(value, bool):
general_settings["store_prompts_in_spend_logs"] = value
elif isinstance(value, str):
# Case-insensitive string comparison
general_settings["store_prompts_in_spend_logs"] = value.lower() == "true"
else:
# For other types, convert to bool
general_settings["store_prompts_in_spend_logs"] = bool(value)
if "disable_auto_add_proxy_admin_to_teams" in _general_settings:
value = _general_settings["disable_auto_add_proxy_admin_to_teams"]
if isinstance(value, str):
general_settings["disable_auto_add_proxy_admin_to_teams"] = value.lower() == "true"
else:
general_settings["disable_auto_add_proxy_admin_to_teams"] = value if value is None else bool(value)
if "apply_user_budget_to_team_keys" in _general_settings and (
"apply_user_budget_to_team_keys" not in self._yaml_general_settings_keys
):
db_value: Final = _general_settings["apply_user_budget_to_team_keys"]
if isinstance(db_value, str):
general_settings["apply_user_budget_to_team_keys"] = db_value.lower() == "true"
else:
general_settings["apply_user_budget_to_team_keys"] = db_value if db_value is None else bool(db_value)
if "enable_openai_websocket_passthrough" not in self._yaml_general_settings_keys:
general_settings["enable_openai_websocket_passthrough"] = _general_settings.get(
"enable_openai_websocket_passthrough"
)
if "user_api_key_cache_max_size" not in self._yaml_general_settings_keys:
db_cache_max_size: Final = _general_settings.get("user_api_key_cache_max_size")
try:
cache_max_size: Final = ConfigGeneralSettings.model_validate(
MappingProxyType({"user_api_key_cache_max_size": db_cache_max_size})
).user_api_key_cache_max_size
except ValidationError:
verbose_proxy_logger.warning(
"Ignoring invalid general_settings.user_api_key_cache_max_size=%r from the DB", db_cache_max_size
)
else:
if cache_max_size is None:
general_settings.pop("user_api_key_cache_max_size", None)
else:
general_settings["user_api_key_cache_max_size"] = cache_max_size
user_api_key_cache.update_in_memory_max_size(cache_max_size)
## STORE MODEL IN DB ##
if "store_model_in_db" in _general_settings:
value = _general_settings["store_model_in_db"]
if value is None:
pass # Don't change store_model_in_db to None; keep current value
elif isinstance(value, bool):
store_model_in_db = value
elif isinstance(value, str):
store_model_in_db = value.lower() == "true"
else:
store_model_in_db = bool(value)
general_settings["store_model_in_db"] = store_model_in_db
## MAXIMUM SPEND LOGS RETENTION PERIOD ##
if "maximum_spend_logs_retention_period" in _general_settings:
old_value: Final = general_settings.get("maximum_spend_logs_retention_period")
new_value: Final = _general_settings["maximum_spend_logs_retention_period"]
general_settings["maximum_spend_logs_retention_period"] = new_value
# Reschedule cleanup job if value changed (including when set to None)
if old_value != new_value:
await self._reschedule_spend_log_cleanup_job()
if "maximum_autorouter_session_retention_period" in _general_settings:
old_session_value: Final = general_settings.get("maximum_autorouter_session_retention_period")
new_session_value: Final = _general_settings["maximum_autorouter_session_retention_period"]
general_settings["maximum_autorouter_session_retention_period"] = new_session_value
if old_session_value != new_session_value:
await self._reschedule_spend_log_cleanup_job()
if "maximum_health_check_retention_period" in _general_settings:
old_health_check_value: Final = general_settings.get("maximum_health_check_retention_period")
new_health_check_value: Final = _general_settings["maximum_health_check_retention_period"]
general_settings["maximum_health_check_retention_period"] = new_health_check_value
if old_health_check_value != new_health_check_value:
await self._reschedule_spend_log_cleanup_job()
## SPEND LOG CLEANUP BOUNDS ##
# The dashboard writes these straight to the DB, so without copying them
# here the running cleanup job never sees them. A key the DB no longer
# carries was cleared from the dashboard, and falls back to whatever
# config.yaml declared, or to None (the shipped default) when it declared
# nothing. Leaving the deleted DB value in memory would keep enforcing the
# bound the operator just removed.
for cleanup_key in SPEND_LOG_CLEANUP_BOUND_SETTINGS:
general_settings[cleanup_key] = _general_settings.get(
cleanup_key, self._yaml_spend_log_cleanup_bounds.get(cleanup_key)
)
for key in (
"user_url_allowed_hosts",
"user_url_validation",
"provider_url_destination_allowed_hosts",
):
if key in _general_settings:
general_settings[key] = _general_settings[key]
_apply_ssrf_general_settings(_general_settings)
def _update_config_fields(
async def _apply_general_settings_side_effects(
self,
current_config: dict,
param_name: Literal[
"general_settings",
"router_settings",
"litellm_settings",
"environment_variables",
],
db_param_value: Any,
) -> dict:
"""
Updates the config fields with the new values from the DB
db_values: Mapping[str, SettingsJsonValue],
cache_size_was_db: bool,
previous_retention_values: tuple[SettingsJsonValue | None, ...],
) -> None:
effects: Final = (
self._apply_alerting_settings,
self._apply_pass_through_settings,
self._apply_boolean_settings,
partial(self._apply_cache_size_setting, cache_size_was_db=cache_size_was_db),
self._apply_store_model_in_db_setting,
partial(self._apply_retention_settings, previous_retention_values=previous_retention_values),
self._apply_ssrf_settings,
)
for effect in effects:
await effect(db_values)
Args:
current_config (dict): Current configuration dictionary to update
param_name (Literal): Name of the parameter to update
db_param_value (Any): New value from the database
async def _apply_alerting_settings(self, db_values: Mapping[str, SettingsJsonValue]) -> None:
alerting: Final = self.settings.get("alerting")
if "alerting" in db_values and isinstance(alerting, list):
proxy_logging_obj.update_values(alerting=alerting)
Returns:
dict: Updated configuration dictionary
"""
alerting_args: Final = self.settings.get("alerting_args")
if "alerting_args" in db_values and self.settings.source("alerting_args") == "db":
proxy_logging_obj.slack_alerting_instance.update_values(alerting_args=alerting_args)
def _deep_merge_dicts(dst: dict, src: dict) -> None:
"""
Deep-merge src into dst, skipping None values and empty lists from src.
On conflicts, src (DB) wins, but empty lists are treated as "no value" and don't overwrite.
"""
stack: Final = [(dst, src)]
while stack:
d, s = stack.pop()
for k, v in s.items():
if v is None:
# Preserve existing config when DB value is None (matches prior behavior)
continue
# Skip empty lists - treat them as "no value" to preserve file config
if isinstance(v, list) and len(v) == 0:
continue
if isinstance(v, dict) and isinstance(d.get(k), dict):
stack.append((d[k], v))
else:
d[k] = v
alert_types: Final = self.settings.get("alert_types")
if "alert_types" in db_values and self.settings.source("alert_types") == "db":
proxy_logging_obj.alert_types = alert_types
proxy_logging_obj.slack_alerting_instance.update_values(alert_types=alert_types, llm_router=llm_router)
# Strip remote-URL module loads from the DB-overlay before merge —
# the YAML-load callsites have ``config_file_path`` set, so a
# DB-sourced ``s3://`` value would otherwise reach
# ``_load_instance_from_remote_storage`` without going through
# the runtime gate.
db_param_value = _scrub_db_overlay_remote_module_loads(section=param_name, db_value=db_param_value)
webhook_url: Final = self.settings.get("alert_to_webhook_url")
if "alert_to_webhook_url" in db_values and self.settings.source("alert_to_webhook_url") == "db":
proxy_logging_obj.slack_alerting_instance.update_values(
alert_to_webhook_url=webhook_url, llm_router=llm_router
)
if param_name == "environment_variables":
decrypted_env_vars = self._decrypt_and_set_db_env_variables(db_param_value, return_original_value=True)
# Normalize keys when loading from DB so services expecting uppercase
# (e.g. Datadog) can read them even if stored in lowercase.
merged_env_vars: Final[dict] = {}
for key, value in decrypted_env_vars.items():
merged_env_vars[key] = value
upper_key = key.upper()
merged_env_vars[upper_key] = value
os.environ[upper_key] = value
if "plugins" in db_values and self.settings.source("plugins") == "db":
register_plugins_from_config(self.settings)
current_config.setdefault("environment_variables", {}).update(merged_env_vars)
return current_config
elif param_name == "litellm_settings" and isinstance(db_param_value, dict):
for key, value in db_param_value.items():
if key in LITELLM_SETTINGS_SAFE_DB_OVERRIDES: # params that are safe to override with db values
setattr(litellm, key, value)
async def _apply_pass_through_settings(self, db_values: Mapping[str, SettingsJsonValue]) -> None:
resolved_endpoints: Final = self.settings.get("pass_through_endpoints")
if "pass_through_endpoints" in db_values and isinstance(resolved_endpoints, list):
await initialize_pass_through_endpoints(pass_through_endpoints=resolved_endpoints)
# If param doesn't exist in config, add it
if param_name not in current_config:
current_config[param_name] = db_param_value
async def _apply_boolean_settings(self, db_values: Mapping[str, SettingsJsonValue]) -> None:
for key in (
"store_prompts_in_spend_logs",
"disable_auto_add_proxy_admin_to_teams",
"apply_user_budget_to_team_keys",
):
if key in db_values and (value := self.settings.get(key)) is not None:
self.settings[key] = coerce_bool(value)
return current_config
# For dictionary values, update only non-none values
if isinstance(current_config[param_name], dict) and isinstance(db_param_value, dict):
_deep_merge_dicts(current_config[param_name], db_param_value)
async def _apply_cache_size_setting(
self,
db_values: Mapping[str, SettingsJsonValue],
cache_size_was_db: bool,
) -> None:
if "user_api_key_cache_max_size" not in db_values and not cache_size_was_db:
return
cache_value: Final = self.settings.get("user_api_key_cache_max_size")
try:
cache_max_size: Final = ConfigGeneralSettings.model_validate(
MappingProxyType({"user_api_key_cache_max_size": cache_value})
).user_api_key_cache_max_size
except ValidationError:
self.settings.pop("user_api_key_cache_max_size", None)
verbose_proxy_logger.warning(
"Ignoring invalid general_settings.user_api_key_cache_max_size=%r from the DB", cache_value
)
return
if cache_max_size is None:
self.settings.pop("user_api_key_cache_max_size", None)
else:
# Non-dict or mismatched types: DB value replaces config (unchanged behavior)
current_config[param_name] = db_param_value
self.settings["user_api_key_cache_max_size"] = cache_max_size
user_api_key_cache.update_in_memory_max_size(cache_max_size)
return current_config
async def _apply_store_model_in_db_setting(self, db_values: Mapping[str, SettingsJsonValue]) -> None:
global store_model_in_db
if "store_model_in_db" not in db_values:
return
value: Final = self.settings.get("store_model_in_db")
if value is None:
return
normalized: Final = coerce_bool(value)
store_model_in_db = normalized if isinstance(normalized, bool) else bool(normalized)
self.settings["store_model_in_db"] = store_model_in_db
async def _apply_retention_settings(
self,
db_values: Mapping[str, SettingsJsonValue],
previous_retention_values: tuple[SettingsJsonValue | None, ...],
) -> None:
if (
any(
key in db_values
for key in (
"maximum_spend_logs_retention_period",
"maximum_autorouter_session_retention_period",
"maximum_health_check_retention_period",
)
)
and previous_retention_values != self._resolved_retention_values()
):
await self._reschedule_spend_log_cleanup_job()
async def _apply_ssrf_settings(self, db_values: Mapping[str, SettingsJsonValue]) -> None:
_apply_ssrf_general_settings(db_values)
async def _update_config_from_db(
self,
@ -7471,37 +7291,48 @@ class ProxyConfig:
verbose_proxy_logger.info("'store_model_in_db' is not True, skipping db updates")
return config
_tasks: Final = []
keys: Final = [
"general_settings",
"router_settings",
"litellm_settings",
"environment_variables",
]
for k in keys:
_tasks.append(get_config_param(prisma_client, k))
responses: Final = await asyncio.gather(*_tasks)
for response in responses:
if response is None:
sections: Final = tuple(self._settings_stores)
responses: Final = await asyncio.gather(*(get_config_param(prisma_client, section) for section in sections))
for section, response in zip(sections, responses):
if response is None or (param_value := getattr(response, "param_value", None)) is None:
continue
param_name = getattr(response, "param_name", None)
param_value = getattr(response, "param_value", None)
verbose_proxy_logger.debug(
"param_name=%s, param_value=%s",
param_name,
_redact_config_param_value_for_logging(param_name, param_value),
section,
_redact_config_param_value_for_logging(section, param_value),
)
if param_name is not None and param_value is not None:
config = self._update_config_fields(
current_config=config,
param_name=param_name,
db_param_value=param_value,
if section == "litellm_settings":
self._apply_litellm_settings_db_values(self._prepared_db_settings_values(section, param_value))
else:
self._settings_stores[section].apply_db_row(
section,
self._prepared_db_settings_values(section, param_value),
)
return config
return self._config_with_resolved_settings(config)
def _prepared_db_settings_values(self, section: Section, value: object) -> Mapping[str, SettingsJsonValue]:
if section == "environment_variables":
decrypted: Final = self._decrypt_and_set_db_env_variables(
dict(_as_settings_mapping(value)), return_original_value=True
)
normalized: Final = {
**decrypted,
**{key.upper(): decrypted_value for key, decrypted_value in decrypted.items()},
}
for key, decrypted_value in normalized.items():
os.environ[key] = decrypted_value
return _as_settings_mapping(normalized)
scrubbed: Final = _scrub_db_overlay_remote_module_loads(section=section, db_value=value)
return _as_settings_mapping(scrubbed)
def _apply_litellm_settings_db_values(self, db_values: Mapping[str, SettingsJsonValue]) -> None:
self.litellm_settings.apply_db_row("litellm_settings", db_values)
for key in LITELLM_SETTINGS_SAFE_DB_OVERRIDES:
if key in db_values and (value := self.litellm_settings.get(key)) is not None:
setattr(litellm, key, value)
def _should_load_db_object(self, object_type: str | SupportedDBObjectType) -> bool:
return should_load_db_object(object_type=object_type)
@ -7737,12 +7568,8 @@ class ProxyConfig:
if config_record is None or config_record.param_value is None:
return
raw_settings: Final = config_record.param_value
litellm_settings: Final = json.loads(raw_settings) if isinstance(raw_settings, str) else raw_settings
if not isinstance(litellm_settings, dict):
return
for key, value in litellm_settings.items():
if key in LITELLM_SETTINGS_SAFE_DB_OVERRIDES:
setattr(litellm, key, value)
db_values: Final = self._prepared_db_settings_values("litellm_settings", raw_settings)
self._apply_litellm_settings_db_values(db_values)
async def _init_semantic_filter_settings_in_db(self, prisma_client: PrismaClient):
"""

5
litellm/proxy/ui_crud_endpoints/proxy_setting_endpoints.py
View file

@ -1483,10 +1483,13 @@ _UI_SETTINGS_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
def apply_runtime_general_settings_flags(ui_settings: Mapping[str, JsonValue]) -> Mapping[str, JsonValue]:
"""Copy the UI settings that gate runtime behavior into ``general_settings``. Returns what was applied."""
from litellm.proxy.config_resolvers import SettingsStore
from litellm.proxy.proxy_server import general_settings
flags: Final = {k: ui_settings[k] for k in _RUNTIME_GENERAL_SETTINGS_FLAGS if k in ui_settings}
if flags:
if isinstance(general_settings, SettingsStore):
general_settings.apply_db_row("ui_settings", flags)
elif flags:
general_settings.update(flags)
return MappingProxyType(flags)

184
litellm/repositories/config_repository.py
View file

@ -1,25 +1,18 @@
"""
Config repository for database operations on LiteLLM_Config.
"""Config repository for database operations on LiteLLM_Config."""
This repository handles config reconciliation between database values and
YAML configmap values. DB values override configmap values except for
None values and empty lists.
"""
from __future__ import annotations
import asyncio
import copy
import json
import os
from collections.abc import Mapping, Sequence
from typing import Any, Final, Literal, Protocol, cast
from typing import TYPE_CHECKING, Final, Protocol, cast
from litellm._logging import verbose_proxy_logger
from litellm.proxy.common_utils.encrypt_decrypt_utils import decrypt_value_helper
if TYPE_CHECKING:
from litellm.proxy.utils import PrismaClient
def _decoded_json(raw: str) -> object:
"""Decode a JSON-encoded config row value into an opaque object."""
return json.loads(raw)
return cast(object, json.loads(raw))
class _ConfigRow(Protocol):
@ -40,16 +33,6 @@ class _ConfigTable(Protocol):
async def delete(self, *, where: Mapping[str, str]) -> _ConfigRow | None: ...
class _ConfigDb(Protocol):
@property
def litellm_config(self) -> _ConfigTable: ...
class _PrismaHandle(Protocol):
@property
def db(self) -> _ConfigDb: ...
class ConfigParam:
"""Simple wrapper for config parameter from DB."""
@ -59,27 +42,20 @@ class ConfigParam:
class ConfigRepository:
"""Repository for config database operations with reconciliation support."""
"""Repository for config database operations."""
CONFIG_PARAMS = [
"general_settings",
"router_settings",
"litellm_settings",
"environment_variables",
]
def __init__(self, prisma_client: Any):
self._prisma_client = prisma_client
def __init__(self, prisma_client: PrismaClient | None):
self._prisma_client: Final = prisma_client
@property
def prisma_client(self) -> _PrismaHandle:
def prisma_client(self) -> PrismaClient:
if self._prisma_client is None:
raise RuntimeError("No DB Connected. See - https://docs.litellm.ai/docs/proxy/virtual_keys")
return self._prisma_client
@property
def _config_table(self) -> _ConfigTable:
return self.prisma_client.db.litellm_config
return cast(_ConfigTable, self.prisma_client.db.litellm_config)
@property
def table(self) -> _ConfigTable:
@ -125,141 +101,3 @@ class ConfigRepository:
param_value = _decoded_json(param_value)
result[record.param_name] = param_value
return result
def _deep_merge_dicts(self, dst: dict, src: dict) -> None:
"""Deep-merge src into dst, skipping None values and empty lists from src.
On conflicts, src (DB) wins, but empty lists are treated as "no value"
and don't overwrite the destination.
"""
stack: Final = [(dst, src)]
while stack:
d, s = stack.pop()
for k, v in s.items():
if v is None:
continue
if isinstance(v, list) and len(v) == 0:
continue
if isinstance(v, dict) and isinstance(d.get(k), dict):
stack.append((d[k], v))
else:
d[k] = v
def _decrypt_env_variables(
self, env_vars: Mapping[str, object], return_original_value: bool = True
) -> dict[str, str]:
"""Decrypt environment variables from database."""
decrypted: Final[dict[str, str]] = {}
for key, value in env_vars.items():
if isinstance(value, str):
decrypted_value = decrypt_value_helper(
value=value,
key=key,
exception_type="debug",
return_original_value=return_original_value,
)
if decrypted_value is not None:
decrypted[key] = decrypted_value
else:
decrypted[key] = str(value)
return decrypted
def _normalize_env_variable_keys(self, env_vars: dict[str, str]) -> dict[str, str]:
"""Normalize env variable keys to include both original and uppercase versions."""
normalized: Final[dict[str, str]] = {}
for key, value in env_vars.items():
normalized[key] = value
upper_key = key.upper()
normalized[upper_key] = value
return normalized
def _update_config_fields(
self,
current_config: dict,
param_name: Literal[
"general_settings",
"router_settings",
"litellm_settings",
"environment_variables",
],
db_param_value: Any,
) -> dict:
"""Update config fields with DB values, handling the merge strategy."""
if param_name == "environment_variables":
decrypted_env_vars: Final = self._decrypt_env_variables(db_param_value, return_original_value=True)
merged_env_vars: Final = self._normalize_env_variable_keys(decrypted_env_vars)
for env_key, value in merged_env_vars.items():
os.environ[env_key] = value
current_config.setdefault("environment_variables", {}).update(merged_env_vars)
return current_config
if param_name not in current_config:
current_config[param_name] = db_param_value
return current_config
if isinstance(current_config[param_name], dict) and isinstance(db_param_value, dict):
self._deep_merge_dicts(current_config[param_name], db_param_value)
else:
current_config[param_name] = db_param_value
return current_config
async def reconcile_config(
self,
yaml_config: dict,
store_model_in_db: bool | None = None,
) -> dict:
"""Reconcile config from YAML with database overrides.
This is the main config reconciliation method that loads config params
from the database and merges them with the YAML config. DB values
override YAML values except for None values and empty lists.
Args:
yaml_config: The configuration loaded from YAML file
store_model_in_db: Whether to load config from DB
Returns:
The merged configuration with DB overrides applied
"""
if store_model_in_db is not True:
verbose_proxy_logger.info("'store_model_in_db' is not True, skipping db config reconciliation")
return yaml_config
tasks: Final = [self.get_param(k) for k in self.CONFIG_PARAMS]
responses: Final = await asyncio.gather(*tasks)
config = copy.deepcopy(yaml_config)
for response in responses:
if response is None:
continue
param_name = response.param_name
param_value = response.param_value
verbose_proxy_logger.debug("param_name=%s, param_value=%s", param_name, param_value)
if param_name is not None and param_value is not None:
config = self._update_config_fields(
current_config=config,
param_name=cast(
Literal[
"general_settings",
"router_settings",
"litellm_settings",
"environment_variables",
],
param_name,
),
db_param_value=param_value,
)
return config
async def prefetch_params(self, param_names: list[str]) -> None:
"""Prefetch config params to warm the cache.
This can be called before reconcile_config to ensure all needed
params are loaded in a single batch.
"""
await asyncio.gather(*[self.get_param(k) for k in param_names])

61
tests/proxy_unit_tests/test_proxy_config_unit_test.py
View file

@ -288,68 +288,55 @@ async def test_json_logs_calls_turn_on_json():
class TestYamlStorePromptsDbOverride:
"""
Test that YAML store_prompts_in_spend_logs takes precedence over DB-cached value.
When store_model_in_db=true, LiteLLM persists general_settings to the DB.
On periodic reloads, _update_general_settings() must NOT override
YAML-explicit values with stale DB values.
"""
def _make_proxy_config_with_yaml_keys(self, yaml_keys: set) -> "ProxyConfig":
"""Helper: create ProxyConfig with pre-populated _yaml_general_settings_keys."""
proxy_config = ProxyConfig()
proxy_config._yaml_general_settings_keys = yaml_keys
return proxy_config
@pytest.mark.asyncio
async def test_yaml_value_takes_precedence_over_db(self):
"""When YAML sets store_prompts_in_spend_logs=false, DB value (true) should be ignored."""
proxy_config = self._make_proxy_config_with_yaml_keys({"store_prompts_in_spend_logs"})
proxy_config = ProxyConfig()
proxy_config.settings.load_yaml({"store_prompts_in_spend_logs": False})
test_general_settings = {"store_prompts_in_spend_logs": False}
with mock.patch("litellm.proxy.proxy_server.general_settings", test_general_settings):
with mock.patch("litellm.proxy.proxy_server.general_settings", proxy_config.settings):
await proxy_config._update_general_settings(
db_general_settings={"store_prompts_in_spend_logs": True},
)
assert test_general_settings["store_prompts_in_spend_logs"] is False
from litellm.proxy import proxy_server
assert proxy_server.general_settings["store_prompts_in_spend_logs"] is False
assert proxy_server.general_settings.source("store_prompts_in_spend_logs") == "config"
@pytest.mark.asyncio
async def test_db_value_used_when_yaml_does_not_set_key(self):
"""When YAML does NOT set store_prompts_in_spend_logs, DB value should be used."""
proxy_config = self._make_proxy_config_with_yaml_keys({"master_key", "database_url"})
proxy_config = ProxyConfig()
proxy_config.settings.load_yaml({"master_key": "sk-test"})
test_general_settings = {"master_key": "sk-test"}
with mock.patch("litellm.proxy.proxy_server.general_settings", test_general_settings):
with mock.patch("litellm.proxy.proxy_server.general_settings", proxy_config.settings):
await proxy_config._update_general_settings(
db_general_settings={"store_prompts_in_spend_logs": True},
)
assert test_general_settings["store_prompts_in_spend_logs"] is True
from litellm.proxy import proxy_server
assert proxy_server.general_settings["store_prompts_in_spend_logs"] is True
assert proxy_server.general_settings.source("store_prompts_in_spend_logs") == "db"
@pytest.mark.asyncio
async def test_admin_ui_change_works_when_yaml_omits_key(self):
"""Admin UI change (DB update) should work when YAML doesn't set the key."""
proxy_config = self._make_proxy_config_with_yaml_keys({"master_key"})
proxy_config = ProxyConfig()
proxy_config.settings.load_yaml({"master_key": "sk-test"})
test_general_settings = {"master_key": "sk-test"}
with mock.patch("litellm.proxy.proxy_server.general_settings", test_general_settings):
with mock.patch("litellm.proxy.proxy_server.general_settings", proxy_config.settings):
await proxy_config._update_general_settings(
db_general_settings={"store_prompts_in_spend_logs": True},
)
assert test_general_settings["store_prompts_in_spend_logs"] is True
await proxy_config._update_general_settings(
db_general_settings={"store_prompts_in_spend_logs": False},
)
assert test_general_settings["store_prompts_in_spend_logs"] is False
from litellm.proxy import proxy_server
def test_yaml_general_settings_keys_populated_on_load(self):
"""_yaml_general_settings_keys should be empty on init."""
assert proxy_server.general_settings["store_prompts_in_spend_logs"] is False
assert proxy_server.general_settings.source("store_prompts_in_spend_logs") == "db"
def test_proxy_config_settings_start_unset(self):
proxy_config = ProxyConfig()
assert proxy_config._yaml_general_settings_keys == set()
assert proxy_config.settings.source("store_prompts_in_spend_logs") == "unset"

41
tests/proxy_unit_tests/test_proxy_utils.py
View file

@ -699,18 +699,19 @@ async def test_proxy_config_update_from_db():
param_name: str
param_value: dict
with patch.object(
pc,
"get_generic_data",
new=AsyncMock(
return_value=ReturnValue(
param_name="litellm_settings",
param_value={
"success_callback": "langfuse",
},
)
),
):
async def get_litellm_settings(_: object, section: str) -> ReturnValue | None:
if section != "litellm_settings":
return None
return ReturnValue(
param_name="litellm_settings",
param_value={
"success_callback": "langfuse",
},
)
proxy_config._load_yaml_settings_stores(test_config)
with patch("litellm.proxy.proxy_server.get_config_param", side_effect=get_litellm_settings):
new_config = await proxy_config._update_config_from_db(
prisma_client=pc,
config=test_config,
@ -1090,7 +1091,7 @@ def test_get_team_models():
assert result == ["gpt-4o", "gpt-3.5-turbo", "gpt-4o-mini"]
def test_update_config_fields():
def test_settings_store_preserves_yaml_team_configuration_when_db_value_is_null():
from litellm.proxy.proxy_server import ProxyConfig
proxy_config = ProxyConfig()
@ -1120,13 +1121,10 @@ def test_update_config_fields():
"context_window_fallbacks": [{"gpt-3.5-turbo": ["gpt-3.5-turbo-large"]}],
},
}
updated_config = proxy_config._update_config_fields(**args)
proxy_config.litellm_settings.load_yaml(args["current_config"]["litellm_settings"])
proxy_config.litellm_settings.apply_db_row("litellm_settings", args["db_param_value"])
all_team_config = proxy_config.litellm_settings["default_team_settings"]
print("updated_config", updated_config)
all_team_config = updated_config["litellm_settings"]["default_team_settings"]
# check if team id config returned
print("all_team_config", all_team_config)
team_config = proxy_config._get_team_config(
team_id="c91e32bb-0f2a-4aa1-86c4-307ca2e03ea3", all_teams_config=all_team_config
)
@ -1135,7 +1133,7 @@ def test_update_config_fields():
assert team_config["langfuse_secret"] == "my-fake-secret"
def test_update_config_fields_default_internal_user_params(monkeypatch):
def test_settings_store_applies_default_internal_user_params_from_db(monkeypatch):
from litellm.proxy.proxy_server import ProxyConfig
proxy_config = ProxyConfig()
@ -1153,7 +1151,8 @@ def test_update_config_fields_default_internal_user_params(monkeypatch):
},
},
}
proxy_config._update_config_fields(**args)
db_values = proxy_config._prepared_db_settings_values("litellm_settings", args["db_param_value"])
proxy_config._apply_litellm_settings_db_values(db_values)
assert litellm.default_internal_user_params == {
"user_role": "proxy_admin",

9
tests/test_litellm/proxy/agent_endpoints/test_endpoints.py
View file

@ -1082,11 +1082,10 @@ class _DbBackedProxyConfig:
db_param_value: Final[dict[str, object]] = json.loads(self.stored_litellm_settings_json)
if not db_param_value:
return config
return ProxyConfig()._update_config_fields(
current_config=config,
param_name="litellm_settings",
db_param_value=db_param_value,
)
proxy_config: Final = ProxyConfig()
db_values: Final = proxy_config._prepared_db_settings_values("litellm_settings", db_param_value)
proxy_config._apply_litellm_settings_db_values(db_values)
return {"litellm_settings": dict(proxy_config.litellm_settings.resolved())}
async def save_config(self, new_config: dict[str, dict[str, object]]) -> None:
self.stored_litellm_settings_json = json.dumps(new_config.get("litellm_settings") or {})

16655
tests/test_litellm/proxy/config_resolvers/fixtures/precedence_matrix.json Normal file
View file

File diff suppressed because it is too large Load diff

84
tests/test_litellm/proxy/config_resolvers/test_settings_rules.py
View file

@ -1,11 +1,26 @@
from __future__ import annotations
from typing import Final
import json
from collections.abc import Mapping
from pathlib import Path
from typing import Final, Literal, cast
import pytest
from litellm.proxy.config_resolvers._descriptors import FieldSource
from litellm.proxy.config_resolvers.settings_rules import ABSENT, DUAL_SOURCE_KEYS, KeyRule, Resolved, resolve
from litellm.proxy.config_resolvers.settings_rules import (
ABSENT,
DUAL_SOURCE_KEYS,
Absent,
JsonValue,
KeyRule,
Resolved,
Section,
SettingValue,
_build_dual_source_keys,
resolve,
rule_for,
)
@pytest.mark.parametrize(
@ -97,3 +112,68 @@ def test_resolve_reports_config_db_and_unset_sources() -> None:
)
assert sources == ("config", "db", "unset")
_PRECEDENCE_MATRIX_PATH: Final = Path(__file__).parent / "fixtures" / "precedence_matrix.json"
def _load_precedence_matrix() -> tuple[dict[str, object], ...]:
raw: Final[object] = json.loads(_PRECEDENCE_MATRIX_PATH.read_text())
assert isinstance(raw, dict)
cases: Final[object] = raw.get("cases")
assert isinstance(cases, list)
assert all(isinstance(case, dict) for case in cases)
return tuple(cast(dict[str, object], case) for case in cases)
def _matrix_value(case: Mapping[str, object], source: Literal["config", "db"]) -> SettingValue:
raw_value: Final[object] = case[source]
assert isinstance(raw_value, Mapping)
present: Final[object] = raw_value.get("present")
assert isinstance(present, bool)
if not present:
return ABSENT
return cast(JsonValue, raw_value["value"])
def test_dual_source_key_registry_matches_the_golden_precedence_matrix() -> None:
registry: Final = _build_dual_source_keys()
for case in _load_precedence_matrix():
section: Final[object] = case["section"]
key: Final[object] = case["key"]
rule_kind: Final[object] = case["rule"]
db_row: Final[object] = case["db_row"]
assert isinstance(section, str)
assert isinstance(key, str)
assert isinstance(rule_kind, str)
assert isinstance(db_row, str)
resolved_rule: Final = registry.get((cast(Section, section), key), registry[(cast(Section, section), "*")])
assert resolved_rule.kind == rule_kind
assert resolved_rule.db_row == db_row
@pytest.mark.parametrize("case", _load_precedence_matrix())
def test_resolve_matches_the_golden_precedence_matrix(case: dict[str, object]) -> None:
section: Final[object] = case["section"]
key: Final[object] = case["key"]
rule_kind: Final[object] = case["rule"]
expected: Final[object] = case["expected"]
assert isinstance(section, str)
assert isinstance(key, str)
assert isinstance(rule_kind, str)
assert isinstance(expected, Mapping)
resolved: Final = resolve(
rule_for(cast(Section, section), key),
_matrix_value(case, "config"),
_matrix_value(case, "db"),
)
expected_present: Final[object] = expected["present"]
assert isinstance(expected_present, bool)
assert rule_for(cast(Section, section), key).kind == rule_kind
assert not isinstance(resolved.value, Absent) is expected_present
if expected_present:
assert resolved.value == expected["value"]
assert resolved.source == expected["source"]

95
tests/test_litellm/proxy/config_resolvers/test_settings_store.py
View file

@ -4,6 +4,7 @@ from typing import Final
import pytest
from litellm.proxy.config_resolvers.settings_rules import JsonValue
from litellm.proxy.config_resolvers.settings_store import SettingsStore
@ -38,6 +39,61 @@ def test_settings_store_matches_plain_dict_mapping_operations() -> None:
}
@pytest.mark.parametrize("operation", ("set", "update", "setdefault", "pop", "delete"))
@pytest.mark.parametrize("initial_value", (None, False, 0, [], ""))
def test_settings_store_mapping_operations_match_a_plain_dict(operation: str, initial_value: JsonValue) -> None:
expected: dict[str, JsonValue] = {"value": initial_value}
store: Final = SettingsStore("general_settings")
store["value"] = initial_value
match operation:
case "set":
expected["value"] = "replacement"
store["value"] = "replacement"
case "update":
expected.update({"value": "replacement", "other": initial_value})
store.update({"value": "replacement", "other": initial_value})
case "setdefault":
assert store.setdefault("value", "replacement") == expected.setdefault("value", "replacement")
assert store.setdefault("other", initial_value) == expected.setdefault("other", initial_value)
case "pop":
assert store.pop("value") == expected.pop("value")
case "delete":
del expected["value"]
del store["value"]
case _:
raise AssertionError(f"unexpected operation: {operation}")
assert dict(store) == expected
assert tuple(store) == tuple(expected)
assert len(store) == len(expected)
assert ("value" in store) is ("value" in expected)
def test_settings_store_keeps_unaffected_runtime_values_on_a_db_row_refresh() -> None:
store: Final = SettingsStore("general_settings")
store.load_yaml({"template": "os.environ/SETTING", "changed": "config"})
store.apply_runtime_values({"template": "resolved", "changed": "resolved-config"})
store.apply_db_row("general_settings", {"changed": "database"})
assert store["template"] == "resolved"
assert store["changed"] == "database"
assert store.source("changed") == "db"
def test_settings_store_removes_only_runtime_values_affected_by_a_cleared_db_row() -> None:
store: Final = SettingsStore("general_settings")
store.load_yaml({"template": "os.environ/SETTING"})
store.apply_db_row("ui_settings", {"allow_public_health_readiness_details": True})
store.apply_runtime_values({"template": "resolved", "allow_public_health_readiness_details": True})
store.apply_db_row("ui_settings", {})
assert store["template"] == "resolved"
assert "allow_public_health_readiness_details" not in store
def test_settings_store_preserves_falsy_config_values_and_provenance() -> None:
store: Final = SettingsStore("general_settings")
yaml_values: Final = {"none": None, "false": False, "zero": 0, "empty_list": [], "empty_string": ""}
@ -92,6 +148,45 @@ def test_settings_store_resolved_view_is_read_only() -> None:
assert store["configured"] == "value"
def test_settings_store_omits_a_null_database_overlay_value() -> None:
store: Final = SettingsStore("router_settings")
store.apply_db_row("router_settings", {"fallbacks": None})
assert "fallbacks" not in store
assert dict(store) == {}
assert store.source("fallbacks") == "unset"
def test_settings_store_keeps_an_empty_database_list_without_a_config_value() -> None:
store: Final = SettingsStore("router_settings")
store.apply_db_row("router_settings", {"fallbacks": []})
assert store["fallbacks"] == []
assert store.source("fallbacks") == "db"
@pytest.mark.asyncio
async def test_load_config_returns_and_binds_the_general_settings_store(tmp_path, monkeypatch) -> None:
from litellm.proxy import proxy_server
from litellm.proxy.proxy_server import ProxyConfig
config_path = tmp_path / "config.yaml"
config_path.write_text("model_list: []\ngeneral_settings:\n max_file_size_mb: 5\n")
monkeypatch.setattr(proxy_server, "prisma_client", None)
monkeypatch.setattr(proxy_server, "store_model_in_db", False)
monkeypatch.delenv("LITELLM_CONFIG_BUCKET_NAME", raising=False)
proxy_config: Final = ProxyConfig()
_router, _models, returned_store = await proxy_config.load_config(router=None, config_file_path=str(config_path))
config_state: Final = proxy_config.get_config_state()
assert returned_store is proxy_config.settings
assert proxy_server.general_settings is proxy_config.settings
assert isinstance(config_state["general_settings"], dict)
assert config_state["general_settings"]["max_file_size_mb"] == 5
def test_settings_store_starts_with_an_unset_source() -> None:
store: Final = SettingsStore("general_settings")

8
tests/test_litellm/proxy/management_endpoints/test_cost_tracking_settings.py
View file

@ -733,11 +733,11 @@ class TestBlockRequestsForModelsWithoutPricing:
from litellm.proxy.proxy_server import ProxyConfig
with patch.object(litellm, "block_requests_for_models_without_pricing", False):
ProxyConfig()._update_config_fields(
current_config={},
param_name="litellm_settings",
db_param_value={"block_requests_for_models_without_pricing": True},
proxy_config = ProxyConfig()
db_values = proxy_config._prepared_db_settings_values(
"litellm_settings", {"block_requests_for_models_without_pricing": True}
)
proxy_config._apply_litellm_settings_db_values(db_values)
assert litellm.block_requests_for_models_without_pricing is True

43
tests/test_litellm/proxy/management_endpoints/test_team_default_params.py
View file

@ -24,13 +24,7 @@ from litellm.proxy.management_endpoints.team_endpoints import (
from litellm.proxy.proxy_server import ProxyConfig
# ---------------------------------------------------------------------------
# _update_config_fields: default_team_params loaded from DB on startup
# ---------------------------------------------------------------------------
class TestConfigFieldsDefaultTeamParams:
"""Tests that _update_config_fields applies default_team_params from DB."""
class TestDefaultTeamParamsFromSettingsStore:
def _make_proxy_config(self) -> ProxyConfig:
return ProxyConfig()
@ -50,11 +44,8 @@ class TestConfigFieldsDefaultTeamParams:
}
}
pc._update_config_fields(
current_config={},
param_name="litellm_settings",
db_param_value=db_settings,
)
db_values = pc._prepared_db_settings_values("litellm_settings", db_settings)
pc._apply_litellm_settings_db_values(db_values)
assert litellm.default_team_params == db_settings["default_team_params"]
@ -68,11 +59,9 @@ class TestConfigFieldsDefaultTeamParams:
}
}
result = pc._update_config_fields(
current_config=config,
param_name="litellm_settings",
db_param_value=db_settings,
)
pc.litellm_settings.load_yaml(config["litellm_settings"])
pc.litellm_settings.apply_db_row("litellm_settings", db_settings)
result = {"litellm_settings": dict(pc.litellm_settings.resolved())}
assert result["litellm_settings"]["default_team_params"] == {"max_budget": 100.0}
# Existing keys preserved
@ -83,11 +72,8 @@ class TestConfigFieldsDefaultTeamParams:
monkeypatch.setattr(litellm, "default_team_params", None)
pc = self._make_proxy_config()
pc._update_config_fields(
current_config={},
param_name="litellm_settings",
db_param_value={"cache": True},
)
db_values = pc._prepared_db_settings_values("litellm_settings", {"cache": True})
pc._apply_litellm_settings_db_values(db_values)
assert litellm.default_team_params is None
@ -111,13 +97,11 @@ class TestConfigFieldsDefaultTeamParams:
}
}
result = pc._update_config_fields(
current_config=config,
param_name="litellm_settings",
db_param_value=db_settings,
)
pc.litellm_settings.load_yaml(config["litellm_settings"])
db_values = pc._prepared_db_settings_values("litellm_settings", db_settings)
pc._apply_litellm_settings_db_values(db_values)
merged = result["litellm_settings"]["default_team_params"]
merged = pc.litellm_settings["default_team_params"]
# DB value wins for max_budget
assert merged["max_budget"] == 200.0
# DB adds rpm_limit
@ -125,8 +109,7 @@ class TestConfigFieldsDefaultTeamParams:
# YAML tpm_limit preserved (not in DB)
assert merged["tpm_limit"] == 100
# setattr should have applied the DB value
assert litellm.default_team_params == db_settings["default_team_params"]
assert litellm.default_team_params == merged
# ---------------------------------------------------------------------------

317
tests/test_litellm/proxy/proxy_server/test_proxy_config.py
View file

@ -134,20 +134,14 @@ def test__scrub_db_overlay_remote_module_loads_invalid_non_dict_returns_input():
def test_resolve_complexity_router_plugins_no_plugins_key_is_a_noop():
config: Dict[str, Any] = {"tiers": {"SIMPLE": "gpt-4o-mini"}}
resolve_complexity_router_plugins(
model_name="smart-router", complexity_router_config=config, config_file_path=None
)
resolve_complexity_router_plugins(model_name="smart-router", complexity_router_config=config, config_file_path=None)
assert config == {"tiers": {"SIMPLE": "gpt-4o-mini"}}
def test_resolve_complexity_router_plugins_resolves_dotted_path_to_live_instance(tmp_path):
plugin_file = tmp_path / "my_plugin.py"
plugin_file.write_text(
"class _Plugin:\n"
" async def run(self, context):\n"
" return context\n"
"\n"
"my_plugin_instance = _Plugin()\n"
"class _Plugin:\n async def run(self, context):\n return context\n\nmy_plugin_instance = _Plugin()\n"
)
config: Dict[str, Any] = {"plugins": ["my_plugin.my_plugin_instance"]}
@ -262,9 +256,18 @@ def _custom_prompt_row(model_name: str) -> dict[str, object]:
[
([_heuristic_v2_row("a"), _heuristic_v2_row("b"), _heuristic_v2_row("c", "heuristic")], "heuristic_v2"),
([_custom_tier_row("a"), _custom_tier_row("b"), _heuristic_v2_row("c", "heuristic")], "tier_definitions"),
([_custom_prompt_row("a"), _custom_prompt_row("b"), _heuristic_v2_row("c", "heuristic")], "operator-written classifier prompt"),
([_custom_tier_row("a"), _custom_prompt_row("b"), _heuristic_v2_row("c", "heuristic")], "operator-written classifier prompt"),
([_operator_examples_row("a"), _custom_tier_row("b"), _heuristic_v2_row("c", "heuristic")], "operator-written classifier prompt"),
(
[_custom_prompt_row("a"), _custom_prompt_row("b"), _heuristic_v2_row("c", "heuristic")],
"operator-written classifier prompt",
),
(
[_custom_tier_row("a"), _custom_prompt_row("b"), _heuristic_v2_row("c", "heuristic")],
"operator-written classifier prompt",
),
(
[_operator_examples_row("a"), _custom_tier_row("b"), _heuristic_v2_row("c", "heuristic")],
"operator-written classifier prompt",
),
],
)
def test_validate_auto_router_capability_limits_refuses_to_start_over_the_limit(
@ -339,20 +342,17 @@ async def test_ProxyConfig_load_config_takes_the_classifier_limit_from_the_licen
),
}
config_yaml = _TWO_HEURISTIC_V2_ROUTERS_YAML.replace(
"classifier_type: heuristic_v2\n", f"classifier_type: {classifier_type}\n{forecast_settings.get(classifier_type, '')}"
"classifier_type: heuristic_v2\n",
f"classifier_type: {classifier_type}\n{forecast_settings.get(classifier_type, '')}",
).replace("tiers: {SIMPLE: gpt-4o-mini}", "tiers: {SIMPLE: gpt-4o-mini, REASONING: gpt-4o}")
f.write_text(config_yaml)
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", None)
monkeypatch.setattr("litellm.proxy.proxy_server.store_model_in_db", False)
monkeypatch.delenv("LITELLM_CONFIG_BUCKET_NAME", raising=False)
monkeypatch.setattr(
"litellm.proxy.proxy_server._license_check.auto_router_capability_limit", lambda: license_limit
)
monkeypatch.setattr("litellm.proxy.proxy_server._license_check.auto_router_capability_limit", lambda: license_limit)
if license_limit is None:
router, _model_list, _general_settings = await ProxyConfig().load_config(
router=None, config_file_path=str(f)
)
router, _model_list, _general_settings = await ProxyConfig().load_config(router=None, config_file_path=str(f))
assert router.auto_router_capability_limit is not None
assert router.auto_router_capability_limit() is None
assert sorted(router.complexity_routers) == ["v2-a", "v2-b"]
@ -371,10 +371,12 @@ async def test_ProxyConfig_load_config_router_refuses_a_db_heuristic_v2_router_b
from litellm.types.router import Deployment
f = tmp_path / "c.yaml"
f.write_text(_TWO_HEURISTIC_V2_ROUTERS_YAML.replace(" - model_name: v2-b\n", " - model_name: v1-b\n", 1).replace(
"classifier_type: heuristic_v2\n tiers: {SIMPLE: gpt-4o-mini}\nrouter_settings",
"classifier_type: heuristic\n tiers: {SIMPLE: gpt-4o-mini}\nrouter_settings",
))
f.write_text(
_TWO_HEURISTIC_V2_ROUTERS_YAML.replace(" - model_name: v2-b\n", " - model_name: v1-b\n", 1).replace(
"classifier_type: heuristic_v2\n tiers: {SIMPLE: gpt-4o-mini}\nrouter_settings",
"classifier_type: heuristic\n tiers: {SIMPLE: gpt-4o-mini}\nrouter_settings",
)
)
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", None)
monkeypatch.setattr("litellm.proxy.proxy_server.store_model_in_db", False)
monkeypatch.delenv("LITELLM_CONFIG_BUCKET_NAME", raising=False)
@ -557,9 +559,7 @@ def test_resolve_complexity_router_plugins_leaves_live_classifier_instance_alone
instance = _Classifier()
config: dict[str, Any] = {"classifier_plugin": instance}
resolve_complexity_router_plugins(
model_name="smart-router", complexity_router_config=config, config_file_path=None
)
resolve_complexity_router_plugins(model_name="smart-router", complexity_router_config=config, config_file_path=None)
assert config["classifier_plugin"] is instance
@ -571,11 +571,7 @@ def test_resolve_complexity_router_plugins_leaves_live_classifier_instance_alone
def test_resolve_routing_plugins_resolves_dotted_paths(tmp_path):
plugin_file = tmp_path / "rs_plugin.py"
plugin_file.write_text(
"class _Plugin:\n"
" async def run(self, context):\n"
" return context\n"
"\n"
"rs_plugin_instance = _Plugin()\n"
"class _Plugin:\n async def run(self, context):\n return context\n\nrs_plugin_instance = _Plugin()\n"
)
resolved = resolve_routing_plugins(
@ -1527,7 +1523,7 @@ async def test_ProxyConfig_get_config_missing_file_raises(monkeypatch):
# ProxyConfig._initialize_secret_manager_from_raw_config
# ---------------------------------------------------------------------------
VAULT_SECRET_MANAGER_MODULE = '''
VAULT_SECRET_MANAGER_MODULE = """
import os
from litellm.integrations.custom_secret_manager import CustomSecretManager
@ -1548,7 +1544,7 @@ class VaultSecretManager(CustomSecretManager):
async def async_read_secret(self, secret_name, optional_params=None, timeout=None, **kwargs):
return VAULT.get(secret_name)
'''
"""
VAULT_BACKED_CONFIG = """
model_list:
@ -1649,9 +1645,7 @@ async def test_ProxyConfig_get_config_reuses_an_already_initialized_secret_manag
@pytest.mark.asyncio
async def test_ProxyConfig_get_config_without_key_management_system_leaves_secret_manager_unset(
tmp_path, monkeypatch
):
async def test_ProxyConfig_get_config_without_key_management_system_leaves_secret_manager_unset(tmp_path, monkeypatch):
"""No ``key_management_system`` means no manager, an unresolvable reference stays None, and
nothing is warned about: with no manager there is nothing to have been absent from."""
config_yaml = VAULT_BACKED_CONFIG.replace(" key_management_system: custom\n", "")
@ -1670,9 +1664,7 @@ async def test_ProxyConfig_get_config_without_key_management_system_leaves_secre
@pytest.mark.asyncio
async def test_ProxyConfig_get_config_warns_when_a_reference_is_missing_from_the_secret_manager(
tmp_path, monkeypatch
):
async def test_ProxyConfig_get_config_warns_when_a_reference_is_missing_from_the_secret_manager(tmp_path, monkeypatch):
"""A reference the manager cannot resolve is logged, instead of silently becoming None."""
config_yaml = VAULT_BACKED_CONFIG.replace("MY_PROVIDER_KEY", "NOT_IN_VAULT")
config_file_path = _write_vault_backed_config(tmp_path, monkeypatch, config_yaml)
@ -2109,10 +2101,7 @@ async def test_ProxyConfig_load_config_minimal_yaml(tmp_path, monkeypatch):
async def test_load_config_logs_disabled_budget_reservation_once(tmp_path, monkeypatch, caplog, setting):
config_file = tmp_path / "budget.yaml"
flag = f" disable_budget_reservation: {setting}\n" if setting is not None else ""
config_file.write_text(
"model_list: []\nlitellm_settings: {}\ngeneral_settings:\n"
" master_key: null\n" + flag
)
config_file.write_text("model_list: []\nlitellm_settings: {}\ngeneral_settings:\n master_key: null\n" + flag)
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", None)
monkeypatch.setattr("litellm.proxy.proxy_server.store_model_in_db", False)
monkeypatch.setattr("litellm.constants.budget_reservation_disabled_info_emitted", False)
@ -2123,10 +2112,7 @@ async def test_load_config_logs_disabled_budget_reservation_once(tmp_path, monke
for _ in range(3):
await config.load_config(router=None, config_file_path=str(config_file))
records = [
record for record in caplog.records
if "disable_budget_reservation is enabled" in record.message
]
records = [record for record in caplog.records if "disable_budget_reservation is enabled" in record.message]
assert [record.levelno for record in records] == ([logging.INFO] if setting == "true" else [])
@ -2138,11 +2124,7 @@ async def test_ProxyConfig_load_config_resolves_router_settings_plugins(tmp_path
to `await "some.string".run(context)`."""
plugin_file = tmp_path / "rs_plugin.py"
plugin_file.write_text(
"class _Plugin:\n"
" async def run(self, context):\n"
" return context\n"
"\n"
"rs_plugin_instance = _Plugin()\n"
"class _Plugin:\n async def run(self, context):\n return context\n\nrs_plugin_instance = _Plugin()\n"
)
f = tmp_path / "c.yaml"
f.write_text(
@ -2157,9 +2139,7 @@ async def test_ProxyConfig_load_config_resolves_router_settings_plugins(tmp_path
monkeypatch.setattr("litellm.proxy.proxy_server.store_model_in_db", False)
monkeypatch.delenv("LITELLM_CONFIG_BUCKET_NAME", raising=False)
router, _model_list, _general_settings = await ProxyConfig().load_config(
router=None, config_file_path=str(f)
)
router, _model_list, _general_settings = await ProxyConfig().load_config(router=None, config_file_path=str(f))
assert len(router.routing_plugins) == 1
assert type(router.routing_plugins[0]).__name__ == "_Plugin"
@ -2226,10 +2206,7 @@ async def test_ProxyConfig_load_config_wires_config_reload_interval(tmp_path, mo
f = tmp_path / "c.yaml"
f.write_text(
"model_list: []\n"
"general_settings:\n"
" proxy_config_reload_interval_seconds: 47\n"
"litellm_settings: {}\n"
"model_list: []\ngeneral_settings:\n proxy_config_reload_interval_seconds: 47\nlitellm_settings: {}\n"
)
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", None)
monkeypatch.setattr("litellm.proxy.proxy_server.store_model_in_db", False)
@ -2371,13 +2348,9 @@ async def test_ProxyConfig__init_non_llm_configs_premium_invalid_worker_registry
async def test_ProxyConfig__init_non_llm_configs_worker_registry_requires_premium(monkeypatch):
monkeypatch.setattr("litellm.proxy.proxy_server.premium_user", False)
pc = ProxyConfig()
with pytest.raises(ValueError, match='Trying to use `worker_registry`You must be a LiteLLM') as exc_info:
with pytest.raises(ValueError, match="Trying to use `worker_registry`You must be a LiteLLM") as exc_info:
await pc._init_non_llm_configs(
config={
"worker_registry": [
{"worker_id": "worker-a", "name": "Worker A", "url": "http://localhost:4001"}
]
},
config={"worker_registry": [{"worker_id": "worker-a", "name": "Worker A", "url": "http://localhost:4001"}]},
config_file_path=None,
)
message = str(exc_info.value)
@ -2607,9 +2580,7 @@ def test_ProxyConfig__warn_on_misplaced_jwt_keys_warns_even_when_also_under_gene
def test_ProxyConfig__warn_on_misplaced_jwt_keys_silent_when_correctly_placed():
"""Keys living only under general_settings are valid, so no warning fires."""
result, warnings = _capture_proxy_warnings(
{"general_settings": {"enable_jwt_auth": True, "litellm_jwtauth": {}}}
)
result, warnings = _capture_proxy_warnings({"general_settings": {"enable_jwt_auth": True, "litellm_jwtauth": {}}})
assert result == ()
assert warnings == []
@ -2636,7 +2607,7 @@ def test_ProxyConfig_initialize_secret_manager_none_noop():
def test_ProxyConfig_initialize_secret_manager_invalid_kms_raises():
pc = ProxyConfig()
with pytest.raises(ValueError, match='Invalid Key Management System selected'):
with pytest.raises(ValueError, match="Invalid Key Management System selected"):
pc.initialize_secret_manager(key_management_system="not-a-real-kms")
@ -3141,28 +3112,6 @@ async def test_ProxyConfig__update_llm_router_no_models_smoke(monkeypatch):
assert snapshot == {"raised": False, "called": True, "models": "empty"}
@pytest.mark.asyncio
async def test_ProxyConfig__update_llm_router_bad_proxy_logging_raises(monkeypatch):
pc = ProxyConfig()
async def fake_get_config():
# alerting present + non-list general_settings to trigger the alerting branch.
return {"general_settings": {"alerting": ["slack"]}}
fake_router = MagicMock()
fake_router.update_settings = MagicMock()
monkeypatch.setattr(pc, "get_config", fake_get_config)
monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", fake_router)
monkeypatch.setattr("litellm.proxy.proxy_server.master_key", "sk-x")
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", None)
monkeypatch.setattr("litellm.proxy.proxy_server.general_settings", {"alerting": ["email"]})
monkeypatch.setattr("litellm.proxy.proxy_server.proxy_config", pc)
# Passing None for proxy_logging_obj triggers AttributeError in _add_general_settings_from_db_config
# when it calls proxy_logging_obj.update_values.
with pytest.raises(AttributeError):
await pc._update_llm_router(new_models=[], proxy_logging_obj=None) # type: ignore[arg-type]
# ---------------------------------------------------------------------------
# ProxyConfig._add_callback_from_db_to_in_memory_litellm_callbacks
# ---------------------------------------------------------------------------
@ -3637,43 +3586,6 @@ async def test_ProxyConfig_get_credentials_reads_from_writer_not_replica(monkeyp
reader_inner.litellm_credentialstable.find_many.assert_not_awaited()
# ---------------------------------------------------------------------------
# ProxyConfig._add_general_settings_from_db_config
# ---------------------------------------------------------------------------
def test_ProxyConfig__add_general_settings_from_db_config_merges_alerting():
pc = ProxyConfig()
proxy_logging = MagicMock()
general = {"alerting": ["slack"]}
config_data = {"general_settings": {"alerting": ["email", "slack"]}}
pc._add_general_settings_from_db_config(
config_data=config_data,
general_settings=general,
proxy_logging_obj=proxy_logging,
)
snapshot = {
"alerting": sorted(general["alerting"]),
"logging_called": proxy_logging.update_values.called,
"merged_count": len(general["alerting"]),
}
assert snapshot == {
"alerting": ["email", "slack"],
"logging_called": True,
"merged_count": 2,
}
def test_ProxyConfig__add_general_settings_from_db_config_bad_config_raises():
pc = ProxyConfig()
with pytest.raises(AttributeError):
pc._add_general_settings_from_db_config(
config_data=None, # type: ignore[arg-type]
general_settings={},
proxy_logging_obj=MagicMock(),
)
# ---------------------------------------------------------------------------
# ProxyConfig._reschedule_spend_log_cleanup_job
# ---------------------------------------------------------------------------
@ -3736,7 +3648,9 @@ async def test_ProxyConfig__update_general_settings_updates_health_check_retenti
reschedule = AsyncMock()
monkeypatch.setattr(pc, "_reschedule_spend_log_cleanup_job", reschedule)
await pc._update_general_settings({"maximum_health_check_retention_period": "30d"})
assert settings["maximum_health_check_retention_period"] == "30d"
from litellm.proxy import proxy_server
assert proxy_server.general_settings["maximum_health_check_retention_period"] == "30d"
reschedule.assert_awaited_once()
@ -3790,7 +3704,6 @@ async def test_ProxyConfig__update_general_settings_yaml_max_batch_file_size_mb_
{"max_batch_file_size_mb": 3},
)
pc = ProxyConfig()
pc._yaml_general_settings_keys = {"max_batch_file_size_mb"}
await pc._update_general_settings({"max_batch_file_size_mb": 5})
from litellm.proxy import proxy_server as ps
@ -3807,7 +3720,7 @@ async def test_ProxyConfig__update_general_settings_cleared_db_max_batch_file_si
await pc._update_general_settings({"max_parallel_requests": 1})
from litellm.proxy import proxy_server as ps
assert ps.general_settings.get("max_batch_file_size_mb") is None
assert ps.general_settings.get("max_batch_file_size_mb") == 8
@pytest.mark.asyncio
@ -3827,7 +3740,6 @@ async def test_ProxyConfig__update_general_settings_yaml_allowed_file_extensions
{"allowed_file_extensions": [".pdf"]},
)
pc = ProxyConfig()
pc._yaml_general_settings_keys = {"allowed_file_extensions"}
await pc._update_general_settings({"allowed_file_extensions": [".jsonl"]})
from litellm.proxy import proxy_server as ps
@ -3845,27 +3757,140 @@ async def test_ProxyConfig__update_general_settings_none_input_noop():
await pc._update_general_settings(db_general_settings=12345) # type: ignore[arg-type]
# ---------------------------------------------------------------------------
# ProxyConfig._update_config_fields
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_ProxyConfig__update_general_settings_skips_redundant_retention_reschedule(monkeypatch):
from litellm.proxy import proxy_server
def test_ProxyConfig__update_config_fields_merges_dict():
pc = ProxyConfig()
current = {"general_settings": {"a": 1, "b": 2}}
out = pc._update_config_fields(
current_config=current,
param_name="general_settings",
db_param_value={"b": 3, "c": 4, "d": 5},
reschedule: Final = AsyncMock()
monkeypatch.setattr(proxy_server, "general_settings", {})
monkeypatch.setattr(pc, "_reschedule_spend_log_cleanup_job", reschedule)
await pc._update_general_settings({"maximum_health_check_retention_period": "30d"})
reschedule.assert_awaited_once()
reschedule.reset_mock()
await pc._update_general_settings({"maximum_health_check_retention_period": "30d"})
reschedule.assert_not_awaited()
@pytest.mark.asyncio
async def test_ProxyConfig__update_general_settings_dispatches_every_side_effect_handler(monkeypatch):
pc = ProxyConfig()
handlers: Final = (
("_apply_alerting_settings", AsyncMock()),
("_apply_pass_through_settings", AsyncMock()),
("_apply_boolean_settings", AsyncMock()),
("_apply_store_model_in_db_setting", AsyncMock()),
("_apply_retention_settings", AsyncMock()),
("_apply_ssrf_settings", AsyncMock()),
("_apply_cache_size_setting", AsyncMock()),
)
assert out == {"general_settings": {"a": 1, "b": 3, "c": 4, "d": 5}}
for name, handler in handlers:
monkeypatch.setattr(pc, name, handler)
await pc._apply_general_settings_side_effects({}, False, ())
for name, handler in handlers:
if name == "_apply_cache_size_setting":
handler.assert_awaited_once_with({}, cache_size_was_db=False)
elif name == "_apply_retention_settings":
handler.assert_awaited_once_with({}, previous_retention_values=())
else:
handler.assert_awaited_once_with({})
def test_ProxyConfig__update_config_fields_invalid_param_raises():
@pytest.mark.asyncio
async def test_ProxyConfig__update_general_settings_unrelated_value_fires_no_runtime_effect(monkeypatch):
from litellm.proxy import proxy_server
pc = ProxyConfig()
with pytest.raises(TypeError):
# Missing required arg.
pc._update_config_fields(current_config={}, param_name="general_settings") # type: ignore[call-arg]
initialize_endpoints: Final = AsyncMock()
reschedule: Final = AsyncMock()
cache: Final = MagicMock()
proxy_logging: Final = MagicMock()
monkeypatch.setattr(proxy_server, "general_settings", {})
monkeypatch.setattr(proxy_server, "initialize_pass_through_endpoints", initialize_endpoints)
monkeypatch.setattr(proxy_server, "user_api_key_cache", cache)
monkeypatch.setattr(proxy_server, "proxy_logging_obj", proxy_logging)
monkeypatch.setattr(pc, "_reschedule_spend_log_cleanup_job", reschedule)
await pc._update_general_settings({"unrelated": "value"})
initialize_endpoints.assert_not_awaited()
reschedule.assert_not_awaited()
cache.update_in_memory_max_size.assert_not_called()
proxy_logging.update_values.assert_not_called()
proxy_logging.slack_alerting_instance.update_values.assert_not_called()
@pytest.mark.asyncio
async def test_ProxyConfig__update_config_from_db_resolves_through_settings_stores(monkeypatch):
pc = ProxyConfig()
config = {
"general_settings": {
"max_file_size_mb": 7,
"max_parallel_requests": 3,
"alerting": ["config"],
"pass_through_endpoints": [{"path": "/config"}],
"maximum_spend_logs_cleanup_batch_size": 10,
},
"router_settings": {"fallbacks": ["config"], "num_retries": 1},
}
db_values = {
"general_settings": {
"max_file_size_mb": 9,
"max_parallel_requests": 11,
"alerting": ["db"],
"pass_through_endpoints": [{"path": "/db"}],
"maximum_spend_logs_cleanup_batch_size": None,
},
"router_settings": {"fallbacks": [], "num_retries": 2},
}
async def get_config_param(_, param_name):
value = db_values.get(param_name)
return SimpleNamespace(param_name=param_name, param_value=value) if value is not None else None
monkeypatch.setattr("litellm.proxy.proxy_server.get_config_param", get_config_param)
pc._load_yaml_settings_stores(config)
resolved = await pc._update_config_from_db(MagicMock(), config, store_model_in_db=True)
assert resolved["general_settings"] == {
"max_file_size_mb": 7,
"max_parallel_requests": 11,
"alerting": ["config", "db"],
"pass_through_endpoints": [{"path": "/db"}, {"path": "/config"}],
"maximum_spend_logs_cleanup_batch_size": 10,
}
assert resolved["router_settings"] == {"fallbacks": ["config"], "num_retries": 2}
assert pc.settings.source("max_file_size_mb") == "config"
assert pc.settings.source("max_parallel_requests") == "db"
@pytest.mark.asyncio
async def test_ProxyConfig_add_deployment_continues_after_null_pass_through_endpoints(monkeypatch):
from litellm.proxy import proxy_server
pc = ProxyConfig()
non_llm_initialization = AsyncMock()
monkeypatch.setattr(proxy_server, "general_settings", {})
monkeypatch.setattr(proxy_server, "prefetch_config_params", AsyncMock())
monkeypatch.setattr(
proxy_server,
"get_config_param",
AsyncMock(return_value=SimpleNamespace(param_value={"pass_through_endpoints": None})),
)
monkeypatch.setattr(proxy_server, "sync_ui_settings_to_general_settings", AsyncMock())
monkeypatch.setattr(pc, "_should_load_db_object", lambda *, object_type: False)
monkeypatch.setattr(pc, "get_credentials", AsyncMock())
monkeypatch.setattr(pc, "_init_non_llm_objects_in_db", non_llm_initialization)
await pc.add_deployment(prisma_client=MagicMock(), proxy_logging_obj=MagicMock())
non_llm_initialization.assert_awaited_once()
# ---------------------------------------------------------------------------

20
tests/test_litellm/proxy/test_plugin_routes.py
View file

@ -14,6 +14,8 @@ Covers three bugs:
import asyncio
from unittest.mock import MagicMock
import pytest
from litellm.proxy._types import (
ConfigGeneralSettings,
LitellmUserRoles,
@ -131,16 +133,16 @@ def test_plugin_key_is_never_returned_to_the_browser() -> None:
register_plugins_from_config({})
def test_db_persisted_plugins_load_on_startup() -> None:
"""Plugins saved to DB general_settings must register when the DB config is
merged at startup, not just when present in the YAML file."""
def test_db_persisted_plugins_load_on_startup(monkeypatch: pytest.MonkeyPatch) -> None:
from litellm.proxy import proxy_server
from litellm.proxy.proxy_server import ProxyConfig
register_plugins_from_config({}) # start empty (as if YAML had no plugins)
register_plugins_from_config({})
monkeypatch.setattr(proxy_server, "general_settings", {})
ProxyConfig()._add_general_settings_from_db_config(
config_data={
"general_settings": {
asyncio.run(
ProxyConfig()._update_general_settings(
{
"plugins": [
{
"name": "db-plugin",
@ -149,9 +151,7 @@ def test_db_persisted_plugins_load_on_startup() -> None:
}
]
}
},
general_settings={},
proxy_logging_obj=MagicMock(),
)
)
names = [p["name"] for p in asyncio.run(list_plugins(user_api_key_dict=_admin()))]

98
tests/test_litellm/proxy/test_proxy_server.py
View file

@ -1087,7 +1087,7 @@ async def test_init_mcp_servers_from_db_respects_supported_db_objects(monkeypatc
mock_init.assert_not_awaited()
def test_update_config_fields_deep_merge_db_wins():
def test_settings_store_deep_merge_db_wins():
from litellm.proxy.proxy_server import ProxyConfig
proxy_config = ProxyConfig()
@ -1127,13 +1127,10 @@ def test_update_config_fields_deep_merge_db_wins():
}
}
updated = proxy_config._update_config_fields(
current_config=current_config,
param_name="router_settings",
db_param_value=db_param_value,
)
proxy_config.router_settings.load_yaml(current_config["router_settings"])
proxy_config.router_settings.apply_db_row("router_settings", db_param_value)
rs = updated["router_settings"]
rs = proxy_config.router_settings.resolved()
aliases = rs["model_group_alias"]
# DB wins on conflicts (deep) for existing alias
@ -5990,7 +5987,7 @@ async def test_init_hashicorp_vault_config_override_retries_on_transport_error()
assert reconnect_kwargs["reason"] == "init_hashicorp_vault_config_override_lookup_failure"
def test_update_config_fields_uppercases_env_vars(monkeypatch):
def test_settings_store_uppercases_db_env_vars(monkeypatch):
"""
Ensure environment variables pulled from DB are uppercased when applied so
integrations like Datadog that expect uppercase env keys can read them.
@ -6001,13 +5998,12 @@ def test_update_config_fields_uppercases_env_vars(monkeypatch):
monkeypatch.delenv(key, raising=False)
proxy_config = ProxyConfig()
updated_config = proxy_config._update_config_fields(
current_config={},
param_name="environment_variables",
db_param_value={"dd_api_key": "test-api-key", "dd_site": "us5.datadoghq.com"},
db_values = proxy_config._prepared_db_settings_values(
"environment_variables", {"dd_api_key": "test-api-key", "dd_site": "us5.datadoghq.com"}
)
proxy_config.environment_variables.apply_db_row("environment_variables", db_values)
env_vars = updated_config.get("environment_variables", {})
env_vars = proxy_config.environment_variables.resolved()
assert env_vars["DD_API_KEY"] == "test-api-key"
assert env_vars["DD_SITE"] == "us5.datadoghq.com"
assert os.environ.get("DD_API_KEY") == "test-api-key"
@ -6465,7 +6461,7 @@ def test_get_config_normalizes_string_callbacks(monkeypatch):
def test_deep_merge_dicts_skips_none_and_empty_lists(monkeypatch):
"""
Test that _update_config_fields deep merge skips None values and empty lists.
Test that SettingsStore deep merge skips None values and empty lists.
"""
from litellm.proxy.proxy_server import ProxyConfig
@ -6492,14 +6488,16 @@ def test_deep_merge_dicts_skips_none_and_empty_lists(monkeypatch):
},
}
result = proxy_config._update_config_fields(current_config, "general_settings", db_param_value)
proxy_config.settings.load_yaml(current_config["general_settings"])
proxy_config.settings.apply_db_row("general_settings", db_param_value)
result = proxy_config.settings.resolved()
assert result["general_settings"]["max_parallel_requests"] == 10
assert result["general_settings"]["allowed_models"] == ["gpt-3.5-turbo", "gpt-4"]
assert result["general_settings"]["new_key"] == "new_value"
assert result["general_settings"]["nested"]["key1"] == "updated_value1"
assert result["general_settings"]["nested"]["key2"] == "value2"
assert result["general_settings"]["nested"]["key3"] == "value3"
assert result["max_parallel_requests"] == 10
assert result["allowed_models"] == ["gpt-3.5-turbo", "gpt-4"]
assert result["new_key"] == "new_value"
assert result["nested"]["key1"] == "updated_value1"
assert result["nested"]["key2"] == "value2"
assert result["nested"]["key3"] == "value3"
class TestInvitationEndpoints:
@ -7343,17 +7341,20 @@ async def test_update_general_settings_clears_a_spend_log_cleanup_bound_dropped_
proxy_config = ProxyConfig()
with patch(
"litellm.proxy.proxy_server.general_settings",
{"maximum_spend_logs_cleanup_run_budget": "90s", "maximum_spend_logs_cleanup_batch_timeout": "10s"},
):
with patch("litellm.proxy.proxy_server.general_settings", proxy_config.settings):
await proxy_config._update_general_settings(
db_general_settings={
"maximum_spend_logs_cleanup_run_budget": "90s",
"maximum_spend_logs_cleanup_batch_timeout": "10s",
}
)
await proxy_config._update_general_settings(
db_general_settings={"maximum_spend_logs_cleanup_batch_timeout": "10s"}
)
import litellm.proxy.proxy_server as ps
assert ps.general_settings["maximum_spend_logs_cleanup_run_budget"] is None
assert "maximum_spend_logs_cleanup_run_budget" not in ps.general_settings
assert ps.general_settings["maximum_spend_logs_cleanup_batch_timeout"] == "10s"
@ -7364,9 +7365,9 @@ async def test_update_general_settings_keeps_a_yaml_set_spend_log_cleanup_bound(
from litellm.proxy.proxy_server import ProxyConfig
proxy_config = ProxyConfig()
proxy_config._yaml_spend_log_cleanup_bounds = {"maximum_spend_logs_cleanup_run_budget": "90s"}
proxy_config.settings.load_yaml({"maximum_spend_logs_cleanup_run_budget": "90s"})
with patch("litellm.proxy.proxy_server.general_settings", {"maximum_spend_logs_cleanup_run_budget": "90s"}):
with patch("litellm.proxy.proxy_server.general_settings", proxy_config.settings):
await proxy_config._update_general_settings(db_general_settings={"store_model_in_db": True})
import litellm.proxy.proxy_server as ps
@ -7382,10 +7383,10 @@ async def test_update_general_settings_clearing_a_db_override_falls_back_to_the_
from litellm.proxy.proxy_server import ProxyConfig
proxy_config = ProxyConfig()
proxy_config._yaml_spend_log_cleanup_bounds = {"maximum_spend_logs_cleanup_run_budget": "90s"}
proxy_config.settings.load_yaml({"maximum_spend_logs_cleanup_run_budget": "90s"})
# Memory currently holds the dashboard override, and the DB no longer carries it.
with patch("litellm.proxy.proxy_server.general_settings", {"maximum_spend_logs_cleanup_run_budget": "30s"}):
with patch("litellm.proxy.proxy_server.general_settings", proxy_config.settings):
await proxy_config._update_general_settings(db_general_settings={"maximum_spend_logs_cleanup_run_budget": "30s"})
await proxy_config._update_general_settings(db_general_settings={"store_model_in_db": True})
import litellm.proxy.proxy_server as ps
@ -7399,9 +7400,9 @@ async def test_update_general_settings_apply_user_budget_to_team_keys_yaml_wins(
from litellm.proxy.proxy_server import ProxyConfig
proxy_config = ProxyConfig()
proxy_config._yaml_general_settings_keys = {"apply_user_budget_to_team_keys"}
proxy_config.settings.load_yaml({"apply_user_budget_to_team_keys": True})
with patch("litellm.proxy.proxy_server.general_settings", {"apply_user_budget_to_team_keys": True}):
with patch("litellm.proxy.proxy_server.general_settings", proxy_config.settings):
await proxy_config._update_general_settings(db_general_settings={"apply_user_budget_to_team_keys": False})
import litellm.proxy.proxy_server as ps
@ -7522,10 +7523,11 @@ async def test_update_general_settings_clearing_user_api_key_cache_max_size_rest
from litellm.proxy.proxy_server import ProxyConfig
cache = UserApiKeyCache()
cache.update_in_memory_max_size(5000)
monkeypatch.setattr(proxy_server_module, "general_settings", {"user_api_key_cache_max_size": 5000})
proxy_config = ProxyConfig()
monkeypatch.setattr(proxy_server_module, "general_settings", proxy_config.settings)
monkeypatch.setattr(proxy_server_module, "user_api_key_cache", cache)
await ProxyConfig()._update_general_settings(db_general_settings={"store_model_in_db": True})
await proxy_config._update_general_settings(db_general_settings={"user_api_key_cache_max_size": 5000})
await proxy_config._update_general_settings(db_general_settings={"store_model_in_db": True})
assert "user_api_key_cache_max_size" not in proxy_server_module.general_settings
@ -7560,10 +7562,10 @@ async def test_update_general_settings_user_api_key_cache_max_size_yaml_wins(mon
from litellm.proxy.proxy_server import ProxyConfig
proxy_config = ProxyConfig()
proxy_config._yaml_general_settings_keys = {"user_api_key_cache_max_size"}
proxy_config.settings.load_yaml({"user_api_key_cache_max_size": 300})
cache = UserApiKeyCache()
cache.update_in_memory_max_size(300)
monkeypatch.setattr(proxy_server_module, "general_settings", {"user_api_key_cache_max_size": 300})
monkeypatch.setattr(proxy_server_module, "general_settings", proxy_config.settings)
monkeypatch.setattr(proxy_server_module, "user_api_key_cache", cache)
await proxy_config._update_general_settings(db_general_settings={"user_api_key_cache_max_size": 10})
@ -7596,7 +7598,10 @@ async def test_update_general_settings_disable_auto_add_proxy_admin_to_teams(db_
import litellm.proxy.proxy_server as ps
assert ps.general_settings["disable_auto_add_proxy_admin_to_teams"] is expected
if expected is None:
assert "disable_auto_add_proxy_admin_to_teams" not in ps.general_settings
else:
assert ps.general_settings["disable_auto_add_proxy_admin_to_teams"] is expected
@pytest.mark.asyncio
@ -11064,11 +11069,8 @@ def test_prompt_caching_settings_propagate_on_config_reload(monkeypatch, field_n
monkeypatch.setattr(litellm, field_name, False if isinstance(db_value, bool) else None)
pc = ps.ProxyConfig()
pc._update_config_fields(
current_config={"litellm_settings": {}},
param_name="litellm_settings",
db_param_value={field_name: db_value},
)
resolved_db_values = pc._prepared_db_settings_values("litellm_settings", {field_name: db_value})
pc._apply_litellm_settings_db_values(resolved_db_values)
assert getattr(litellm, field_name) == db_value
@ -13782,8 +13784,8 @@ def test_disabling_docs_does_not_disable_other_routes(monkeypatch):
"db_general_settings, expected",
[
({"enable_openai_websocket_passthrough": True}, True),
({"enable_openai_websocket_passthrough": False}, False),
({}, None),
({"enable_openai_websocket_passthrough": False}, True),
({}, True),
],
)
async def test_update_general_settings_propagates_openai_websocket_passthrough(db_general_settings, expected):
@ -13804,9 +13806,9 @@ async def test_update_general_settings_keeps_yaml_openai_websocket_passthrough()
from litellm.proxy.proxy_server import ProxyConfig
proxy_config = ProxyConfig()
proxy_config._yaml_general_settings_keys = {"enable_openai_websocket_passthrough"}
proxy_config.settings.load_yaml({"enable_openai_websocket_passthrough": False})
with patch("litellm.proxy.proxy_server.general_settings", {"enable_openai_websocket_passthrough": False}):
with patch("litellm.proxy.proxy_server.general_settings", proxy_config.settings):
await proxy_config._update_general_settings(db_general_settings={"enable_openai_websocket_passthrough": True})
import litellm.proxy.proxy_server as ps

14
tests/test_litellm/proxy/ui_crud_endpoints/test_proxy_setting_endpoints.py
View file

@ -3439,3 +3439,17 @@ class TestSyncUiSettingsToGeneralSettings:
assert dict(applied) == {}
assert general_settings == {"allow_agents_for_team_admins": True}
def test_applied_runtime_flags_keep_the_ui_row_as_the_source(self, monkeypatch):
from litellm.proxy import proxy_server
from litellm.proxy.config_resolvers import SettingsStore
from litellm.proxy.ui_crud_endpoints.proxy_setting_endpoints import apply_runtime_general_settings_flags
general_settings = SettingsStore("general_settings")
general_settings.load_yaml({"forward_client_headers_to_llm_api": False})
monkeypatch.setattr(proxy_server, "general_settings", general_settings)
apply_runtime_general_settings_flags({"forward_client_headers_to_llm_api": True})
assert general_settings["forward_client_headers_to_llm_api"] is True
assert general_settings.source("forward_client_headers_to_llm_api") == "db"

156
tests/test_litellm/repositories/test_repositories.py
View file

@ -1447,27 +1447,6 @@ class TestConfigRepository:
client = MockPrismaClient()
return ConfigRepository(client)
def test_deep_merge_dicts_db_wins(self, repo):
dst = {"a": 1, "b": {"c": 2}}
src = {"a": 10, "b": {"d": 3}}
repo._deep_merge_dicts(dst, src)
assert dst["a"] == 10
assert dst["b"]["c"] == 2
assert dst["b"]["d"] == 3
def test_deep_merge_dicts_skips_none(self, repo):
dst = {"a": 1}
src = {"a": None, "b": 2}
repo._deep_merge_dicts(dst, src)
assert dst["a"] == 1
assert dst["b"] == 2
def test_deep_merge_dicts_skips_empty_list(self, repo):
dst = {"models": ["gpt-4"]}
src = {"models": []}
repo._deep_merge_dicts(dst, src)
assert dst["models"] == ["gpt-4"]
@pytest.mark.asyncio
async def test_get_param(self, repo):
repo._prisma_client.db.litellm_config._records["general_settings"] = {
@ -1512,99 +1491,6 @@ class TestConfigRepository:
params = await repo.get_all_params()
assert len(params) == 2
@pytest.mark.asyncio
async def test_reconcile_config_skips_when_store_model_false(self, repo):
yaml_config = {"general_settings": {"key": "value"}}
result = await repo.reconcile_config(yaml_config, store_model_in_db=False)
assert result == yaml_config
@pytest.mark.asyncio
async def test_prefetch_params(self, repo):
repo._prisma_client.db.litellm_config._records["general_settings"] = {
"param_name": "general_settings",
"param_value": "{}",
}
await repo.prefetch_params(["general_settings"])
@pytest.mark.asyncio
async def test_reconcile_config_with_db_values(self, repo):
repo._prisma_client.db.litellm_config._records["general_settings"] = {
"param_name": "general_settings",
"param_value": '{"master_key": "db-key", "db_only": "from_db"}',
}
repo._prisma_client.db.litellm_config._records["router_settings"] = {
"param_name": "router_settings",
"param_value": '{"timeout": 60}',
}
yaml_config = {
"general_settings": {"master_key": "yaml-key", "yaml_only": "from_yaml"},
}
result = await repo.reconcile_config(yaml_config, store_model_in_db=True)
assert result["general_settings"]["master_key"] == "db-key"
assert result["general_settings"]["yaml_only"] == "from_yaml"
assert result["general_settings"]["db_only"] == "from_db"
assert result["router_settings"]["timeout"] == 60
@pytest.mark.asyncio
@patch("litellm.repositories.config_repository.decrypt_value_helper")
async def test_reconcile_config_with_environment_variables(
self, mock_decrypt, repo
):
mock_decrypt.side_effect = lambda value, **kw: f"decrypted_{value}"
repo._prisma_client.db.litellm_config._records["environment_variables"] = {
"param_name": "environment_variables",
"param_value": '{"api_key": "encrypted_key", "secret": "encrypted_secret"}',
}
yaml_config = {}
result = await repo.reconcile_config(yaml_config, store_model_in_db=True)
assert "environment_variables" in result
assert "api_key" in result["environment_variables"]
assert "API_KEY" in result["environment_variables"]
@pytest.mark.asyncio
async def test_reconcile_config_none_values_preserved(self, repo):
repo._prisma_client.db.litellm_config._records["general_settings"] = {
"param_name": "general_settings",
"param_value": '{"new_key": "value", "null_key": null}',
}
yaml_config = {"general_settings": {"existing": "keep"}}
result = await repo.reconcile_config(yaml_config, store_model_in_db=True)
assert result["general_settings"]["existing"] == "keep"
assert result["general_settings"]["new_key"] == "value"
def test_update_config_fields_non_dict(self, repo):
config = {"litellm_settings": "old_value"}
result = repo._update_config_fields(
current_config=config,
param_name="litellm_settings",
db_param_value="new_value",
)
assert result["litellm_settings"] == "new_value"
def test_update_config_fields_new_param(self, repo):
config = {}
result = repo._update_config_fields(
current_config=config,
param_name="router_settings",
db_param_value={"timeout": 30},
)
assert result["router_settings"] == {"timeout": 30}
@patch("litellm.repositories.config_repository.decrypt_value_helper")
def test_decrypt_env_variables_non_string(self, mock_decrypt, repo):
mock_decrypt.side_effect = lambda value, **kw: value
env_vars = {"string_val": "encrypted", "int_val": 123, "bool_val": True}
result = repo._decrypt_env_variables(env_vars)
assert result["int_val"] == "123"
assert result["bool_val"] == "True"
@patch("litellm.repositories.config_repository.decrypt_value_helper")
def test_decrypt_env_variables_none_value(self, mock_decrypt, repo):
mock_decrypt.return_value = None
env_vars = {"key": "value"}
result = repo._decrypt_env_variables(env_vars)
assert "key" not in result
class TestVerificationTokenRepositoryExtended:
@pytest.fixture
@ -2213,48 +2099,6 @@ class TestTeamRepositoryArchiveData:
assert "router_settings" in archive_data
class TestConfigRepositoryDeepCopy:
@pytest.fixture
def repo(self):
client = MockPrismaClient()
return ConfigRepository(client)
@pytest.mark.asyncio
async def test_reconcile_config_does_not_mutate_original(self, repo):
import copy
repo._prisma_client.db.litellm_config._records["general_settings"] = {
"param_name": "general_settings",
"param_value": '{"db_key": "db_value", "nested": {"db_nested": "from_db"}}',
}
original_config = {
"general_settings": {
"yaml_key": "yaml_value",
"nested": {"yaml_nested": "from_yaml"},
}
}
original_copy = copy.deepcopy(original_config)
result = await repo.reconcile_config(original_config, store_model_in_db=True)
assert original_config == original_copy
assert result["general_settings"]["db_key"] == "db_value"
assert result["general_settings"]["yaml_key"] == "yaml_value"
assert result["general_settings"]["nested"]["db_nested"] == "from_db"
assert result["general_settings"]["nested"]["yaml_nested"] == "from_yaml"
@pytest.mark.asyncio
async def test_reconcile_config_repeated_calls_independent(self, repo):
repo._prisma_client.db.litellm_config._records["general_settings"] = {
"param_name": "general_settings",
"param_value": '{"db_key": "db_value"}',
}
yaml_config = {"general_settings": {"yaml_key": "yaml_value"}}
result1 = await repo.reconcile_config(yaml_config, store_model_in_db=True)
result1["general_settings"]["modified"] = "in_result1"
result2 = await repo.reconcile_config(yaml_config, store_model_in_db=True)
assert "modified" not in yaml_config.get("general_settings", {})
assert "modified" not in result2.get("general_settings", {})
class TestPrismaTableRepository:
def test_table_property_returns_named_delegate(self):
from litellm.proxy.common_utils.config_sync_pubsub import (