fix(proxy): keep cache scope types explicit

litellm/caching/caching.py

@@ -388,17 +388,20 @@ class Cache:
 
     def _get_proxy_cache_scope(self, kwargs: Mapping[str, object]) -> str:
         """Return a stable, non-secret scope for authenticated proxy callers."""
-        metadata_sources: Final[tuple[Mapping[str, object], ...]] = (
-            tuple(  # comprehension-ok: inspect both metadata locations
-                source.get(key) or {}
-                for source in (kwargs, kwargs.get("litellm_params") or {})
-                for key in ("metadata", "litellm_metadata")
-            )
-        )
-        scope_values: Final = (
-            (field, next((source[field] for source in metadata_sources if source.get(field) is not None), None))
-            for field in self._PROXY_CACHE_SCOPE_FIELDS
-        )
+
+        def as_mapping(value: object) -> Mapping[str, object]:
+            return value if isinstance(value, Mapping) else {}
+
+        litellm_params = as_mapping(kwargs.get("litellm_params"))
+        metadata_sources: list[Mapping[str, object]] = []
+        for source in (kwargs, litellm_params):
+            for key in ("metadata", "litellm_metadata"):
+                metadata_sources.append(as_mapping(source.get(key)))
+
+        scope_values: list[tuple[str, object | None]] = []
+        for field in self._PROXY_CACHE_SCOPE_FIELDS:
+            value = next((source[field] for source in metadata_sources if source.get(field) is not None), None)
+            scope_values.append((field, value))
         return "".join(f"{field}: {value}" for field, value in scope_values if value is not None)
 
     def get_cache_key(self, **kwargs: object) -> str:
@@ -440,9 +443,9 @@ class Cache:
         if is_semantic_cache:
             cache_key += self._get_semantic_cache_tenant_scope(kwargs)
         else:
-            cache_key += self._get_proxy_cache_scope(kwargs)  # rebind-ok: exact cache key is assembled incrementally
-        else:
-            cache_key += self._get_proxy_cache_scope(kwargs)  # rebind-ok: exact cache key is assembled incrementally
+            cache_key += self._get_proxy_cache_scope(
+                cast(Mapping[str, object], kwargs)
+            )  # cast-ok: cache kwargs are string-keyed
 
         reasoning_summary: Final = peek_reasoning_summary_aliases(kwargs)
         reasoning_summary_material: Final = (

tests/unit/proxy/proxy_server/test_proxy_config.py

@@ -29,9 +29,12 @@ from prisma.errors import DataError
 from pydantic import JsonValue, TypeAdapter, ValidationError
 
 import litellm
+import litellm.proxy.proxy_server as proxy_server
 from litellm.proxy._types import CommonProxyErrors, ConfigGeneralSettings
+from litellm.proxy.utils import PrismaClient
 from litellm.proxy.common_utils.encrypt_decrypt_utils import encrypt_value_helper
 from litellm.types.router import Deployment, LiteLLM_Params
+from litellm import Router
 from litellm.proxy.proxy_server import (
     ProxyConfig,
     _is_remote_module_url,
@@ -191,10 +194,10 @@ async def test_db_router_rebuild_serves_same_proxy_caller_from_cache_and_isolate
     monkeypatch.setattr("litellm.proxy.proxy_server.llm_model_list", [])
     monkeypatch.setattr("litellm.proxy.proxy_server.general_settings", {})
 
-    async def empty_config() -> dict:
+    async def empty_config() -> Mapping[str, object]:
         return {}
 
-    async def no_router_settings(_router: object, _prisma: object) -> None:
+    async def no_router_settings(_router: Router | None, _prisma: PrismaClient | None) -> None:
         return None
 
     await proxy_config._update_llm_router(
@@ -204,7 +207,8 @@ async def test_db_router_rebuild_serves_same_proxy_caller_from_cache_and_isolate
         model_decryptor=lambda _models: fake_model_list,
         router_settings_loader=no_router_settings,
     )
-    router = __import__("litellm.proxy.proxy_server", fromlist=["llm_router"]).llm_router
+    router = proxy_server.llm_router
+    assert router is not None
 
     request: Final = [{"role": "user", "content": "cache canary"}]
     caller_a: Final = {"user_api_key_hash": "hash-a"}
@@ -238,10 +242,10 @@ async def test_db_router_rebuild_cache_disabled_does_not_reuse_responses(
     monkeypatch.setattr("litellm.proxy.proxy_server.llm_model_list", [])
     monkeypatch.setattr("litellm.proxy.proxy_server.general_settings", {})
 
-    async def empty_config() -> dict:
+    async def empty_config() -> Mapping[str, object]:
         return {}
 
-    async def no_router_settings(_router: object, _prisma: object) -> None:
+    async def no_router_settings(_router: Router | None, _prisma: PrismaClient | None) -> None:
         return None
 
     await proxy_config._update_llm_router(
@@ -251,7 +255,8 @@ async def test_db_router_rebuild_cache_disabled_does_not_reuse_responses(
         model_decryptor=lambda _models: fake_model_list,
         router_settings_loader=no_router_settings,
     )
-    router = __import__("litellm.proxy.proxy_server", fromlist=["llm_router"]).llm_router
+    router = proxy_server.llm_router
+    assert router is not None
     request: Final = [{"role": "user", "content": "uncached canary"}]
     first = await router.acompletion(model="fake-model", messages=request)
     second = await router.acompletion(model="fake-model", messages=request)
@@ -271,7 +276,6 @@ async def test_proxy_config_loads_lens_store_from_yaml(tmp_path, monkeypatch) ->
     assert is_lens_tracing_enabled(settings["tracing"], {}) is True
 
 
-
 @pytest.mark.asyncio
 @pytest.mark.parametrize("shutdown_error", [False, True])
 async def test_tracing_config_automatically_exports_spend_without_a_storage_dependency(
@@ -297,7 +301,9 @@ async def test_tracing_config_automatically_exports_spend_without_a_storage_depe
     outcome: Final = pytest.raises(RuntimeError, match="shutdown failure") if shutdown_error else nullcontext()
     with outcome:
         async with manage_tracing(enabled=True, client_factory=client):
-            logger: Final = next(callback for callback in litellm._async_success_callback if isinstance(callback, LensExporter))
+            logger: Final = next(
+                callback for callback in litellm._async_success_callback if isinstance(callback, LensExporter)
+            )
             await logger.async_log_success_event(
                 {"standard_logging_object": {"id": "response-1", "response_cost": 0.25}}, None, None, None
             )
@@ -5089,7 +5095,9 @@ async def test_ProxyConfig__update_config_from_db_keeps_keys_the_config_file_omi
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize("ui_settings_already_synced", [False, True])
-async def test_ProxyConfig_add_deployment_continues_after_null_pass_through_endpoints(monkeypatch, ui_settings_already_synced):
+async def test_ProxyConfig_add_deployment_continues_after_null_pass_through_endpoints(
+    monkeypatch, ui_settings_already_synced
+):
     from litellm.proxy import proxy_server
 
     pc = ProxyConfig()