use cached keys and teams for router settings

litellm/proxy/_types.py

@@ -2162,6 +2162,7 @@ class LiteLLM_VerificationToken(LiteLLMPydanticObjectBase):
     rotation_interval: Optional[str] = None  # How often to rotate (e.g., "30d", "90d")
     last_rotation_at: Optional[datetime] = None  # When this key was last rotated
     key_rotation_at: Optional[datetime] = None  # When this key should next be rotated
+    router_settings: Optional[dict] = None
     model_config = ConfigDict(protected_namespaces=())
 
 

litellm/proxy/common_request_processing.py

@@ -616,19 +616,21 @@ class ProxyBaseLLMRequestProcessing:
             user_api_key_dict=user_api_key_dict, data=self.data, call_type=route_type  # type: ignore
         )
 
-        # Apply hierarchical router_settings (Key > Team > Global)
+        # Apply hierarchical router_settings (Key > Team)
+        # Global router_settings are already on the Router object itself.
         if llm_router is not None and proxy_config is not None:
             from litellm.proxy.proxy_server import prisma_client
 
             router_settings = await proxy_config._get_hierarchical_router_settings(
                 user_api_key_dict=user_api_key_dict,
                 prisma_client=prisma_client,
+                proxy_logging_obj=proxy_logging_obj,
             )
 
-            # If router_settings found (from key, team, or global), apply them
+            # If router_settings found (from key or team), apply them
             # Pass settings as per-request overrides instead of creating a new Router
             # This avoids expensive Router instantiation on each request
-            if router_settings is not None and router_settings:
+            if router_settings is not None:
                 self.data["router_settings_override"] = router_settings
 
         if "messages" in self.data and self.data["messages"]:

litellm/proxy/proxy_server.py

@@ -3403,6 +3403,86 @@ class ProxyConfig:
             decrypted_variables[k] = decrypted_value
         return decrypted_variables
 
+    @staticmethod
+    def _parse_router_settings_value(value: Any) -> Optional[dict]:
+        """
+        Parse a router_settings value that may be a dict or a JSON/YAML string.
+
+        Returns a non-empty dict if valid, otherwise None.
+        """
+        if value is None:
+            return None
+
+        parsed: Optional[dict] = None
+        if isinstance(value, dict):
+            parsed = value
+        elif isinstance(value, str):
+            import json
+
+            import yaml
+
+            try:
+                parsed = yaml.safe_load(value)
+            except (yaml.YAMLError, json.JSONDecodeError):
+                try:
+                    parsed = json.loads(value)
+                except json.JSONDecodeError:
+                    pass
+
+        if isinstance(parsed, dict) and parsed:
+            return parsed
+        return None
+
+    async def _get_hierarchical_router_settings(
+        self,
+        user_api_key_dict: Optional["UserAPIKeyAuth"],
+        prisma_client: Optional[PrismaClient],
+        proxy_logging_obj: Optional["ProxyLogging"] = None,
+    ) -> Optional[dict]:
+        """
+        Get router_settings in priority order: Key > Team
+
+        Uses the already-cached key object and the cached team lookup
+        (get_team_object) to avoid direct DB queries on the hot path.
+
+        Global router_settings are NOT looked up here — they are already
+        applied to the Router object at config-load / DB-sync time.
+
+        Returns:
+            dict: router_settings, or None if no settings found
+        """
+        # 1. Try key-level router_settings
+        # user_api_key_dict is already the cached/authenticated key object —
+        # no DB call needed.
+        if user_api_key_dict is not None:
+            key_settings = self._parse_router_settings_value(
+                getattr(user_api_key_dict, "router_settings", None)
+            )
+            if key_settings is not None:
+                return key_settings
+
+        # 2. Try team-level router_settings using cached team lookup
+        # get_team_object checks in-memory cache / Redis first, only falls
+        # back to DB on a cache miss.
+        if user_api_key_dict is not None and user_api_key_dict.team_id is not None:
+            try:
+                team_obj = await get_team_object(
+                    team_id=user_api_key_dict.team_id,
+                    prisma_client=prisma_client,
+                    user_api_key_cache=user_api_key_cache,
+                    proxy_logging_obj=proxy_logging_obj,
+                )
+                team_settings = self._parse_router_settings_value(
+                    getattr(team_obj, "router_settings", None)
+                )
+                if team_settings is not None:
+                    return team_settings
+            except Exception:
+                # If team lookup fails, no team-level settings available
+                pass
+
+        return None
+
     async def _add_router_settings_from_db_config(
         self,
         config_data: dict,

tests/test_litellm/proxy/test_common_request_processing.py

@@ -78,7 +78,6 @@ class TestProxyBaseLLMRequestProcessing:
         assert data_passed["litellm_call_id"] == returned_data["litellm_call_id"]
 
     @pytest.mark.asyncio
-<<<<<<< HEAD
     async def test_should_apply_hierarchical_router_settings_as_override(
         self, monkeypatch
     ):
@@ -148,6 +147,7 @@ class TestProxyBaseLLMRequestProcessing:
         mock_proxy_config._get_hierarchical_router_settings.assert_called_once_with(
             user_api_key_dict=mock_user_api_key_dict,
             prisma_client=mock_prisma_client,
+            proxy_logging_obj=mock_proxy_logging_obj,
         )
         # get_model_list should NOT be called - we no longer copy model list for per-request routers
         mock_llm_router.get_model_list.assert_not_called()
@@ -165,8 +165,6 @@ class TestProxyBaseLLMRequestProcessing:
         assert "model_list" not in router_settings_override
 
     @pytest.mark.asyncio
-=======
->>>>>>> origin
     async def test_stream_timeout_header_processing(self):
         """
         Test that x-litellm-stream-timeout header gets processed and added to request data as stream_timeout.