Fix ruff PLR0915 lint errors (#21766)

* fix(content_filter): extract helpers to reduce __init__ statement count

Fixes PLR0915 ruff lint error (too many statements).

* fix(test_eval): extract print/save helpers to reduce statement count

Fixes PLR0915 ruff lint error (too many statements).

* fix(proxy/utils): extract lock-timeout logic to reduce statement count

Fixes PLR0915 ruff lint error (too many statements).

litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py

@@ -10,8 +10,19 @@ import json
 import os
 import re
 from datetime import datetime
-from typing import (TYPE_CHECKING, Any, AsyncGenerator, Dict, List, Literal,
-                    Optional, Pattern, Tuple, Union, cast)
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    AsyncGenerator,
+    Dict,
+    List,
+    Literal,
+    Optional,
+    Pattern,
+    Tuple,
+    Union,
+    cast,
+)
 
 import yaml
 from fastapi import HTTPException
@@ -20,22 +31,37 @@ from litellm import Router
 from litellm._logging import verbose_proxy_logger
 from litellm.integrations.custom_guardrail import CustomGuardrail
 from litellm.proxy._types import UserAPIKeyAuth
-from litellm.types.utils import (GenericGuardrailAPIInputs, GuardrailStatus,
-                                 GuardrailTracingDetail, ModelResponseStream)
+from litellm.types.utils import (
+    GenericGuardrailAPIInputs,
+    GuardrailStatus,
+    GuardrailTracingDetail,
+    ModelResponseStream,
+)
 
 if TYPE_CHECKING:
     from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
 
-from litellm.types.guardrails import (BlockedWord, ContentFilterAction,
-                                      ContentFilterPattern,
-                                      GuardrailEventHooks, Mode)
+from litellm.types.guardrails import (
+    BlockedWord,
+    ContentFilterAction,
+    ContentFilterPattern,
+    GuardrailEventHooks,
+    Mode,
+)
 from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter import (
-    BlockedWordDetection, CategoryKeywordDetection, CompetitorIntentDetection,
-    CompetitorIntentResult, ContentFilterCategoryConfig,
-    ContentFilterDetection, PatternDetection)
+    BlockedWordDetection,
+    CategoryKeywordDetection,
+    CompetitorIntentDetection,
+    CompetitorIntentResult,
+    ContentFilterCategoryConfig,
+    ContentFilterDetection,
+    PatternDetection,
+)
 
-from .competitor_intent import (AirlineCompetitorIntentChecker,
-                                BaseCompetitorIntentChecker)
+from .competitor_intent import (
+    AirlineCompetitorIntentChecker,
+    BaseCompetitorIntentChecker,
+)
 from .patterns import PATTERN_EXTRA_CONFIG, get_compiled_pattern
 
 MAX_KEYWORD_VALUE_GAP_WORDS = 1
@@ -196,48 +222,15 @@ class ContentFilterGuardrail(CustomGuardrail):
         # Competitor intent checker (optional; airline uses major_airlines.json, generic requires competitors)
         self._competitor_intent_checker: Optional[BaseCompetitorIntentChecker] = None
         if competitor_intent_config and isinstance(competitor_intent_config, dict):
-            try:
-                competitor_intent_type = competitor_intent_config.get(
-                    "competitor_intent_type", "airline"
-                )
-                if competitor_intent_type == "generic":
-                    self._competitor_intent_checker = BaseCompetitorIntentChecker(
-                        competitor_intent_config
-                    )
-                else:
-                    self._competitor_intent_checker = AirlineCompetitorIntentChecker(
-                        competitor_intent_config
-                    )
-                verbose_proxy_logger.debug(
-                    "ContentFilterGuardrail: competitor intent checker enabled (%s)",
-                    competitor_intent_type,
-                )
-            except Exception as e:
-                verbose_proxy_logger.warning(
-                    "ContentFilterGuardrail: failed to init competitor intent checker: %s",
-                    e,
-                )
+            self._init_competitor_intent_checker(competitor_intent_config)
 
         # Load categories if provided
         if categories:
             self._load_categories(categories)
 
         # Normalize inputs: convert dicts to Pydantic models for consistent handling
-        normalized_patterns: List[ContentFilterPattern] = []
-        if patterns:
-            for pattern_config in patterns:
-                if isinstance(pattern_config, dict):
-                    normalized_patterns.append(ContentFilterPattern(**pattern_config))
-                else:
-                    normalized_patterns.append(pattern_config)
-
-        normalized_blocked_words: List[BlockedWord] = []
-        if blocked_words:
-            for word in blocked_words:
-                if isinstance(word, dict):
-                    normalized_blocked_words.append(BlockedWord(**word))
-                else:
-                    normalized_blocked_words.append(word)
+        normalized_patterns = self._normalize_patterns(patterns)
+        normalized_blocked_words = self._normalize_blocked_words(blocked_words)
 
         # Compile regex patterns
         self.compiled_patterns: List[Dict[str, Any]] = []
@@ -278,6 +271,57 @@ class ContentFilterGuardrail(CustomGuardrail):
             f"{len(self.category_keywords)} keywords"
         )
 
+    def _init_competitor_intent_checker(
+        self, competitor_intent_config: Dict[str, Any]
+    ) -> None:
+        try:
+            competitor_intent_type = competitor_intent_config.get(
+                "competitor_intent_type", "airline"
+            )
+            if competitor_intent_type == "generic":
+                self._competitor_intent_checker = BaseCompetitorIntentChecker(
+                    competitor_intent_config
+                )
+            else:
+                self._competitor_intent_checker = AirlineCompetitorIntentChecker(
+                    competitor_intent_config
+                )
+            verbose_proxy_logger.debug(
+                "ContentFilterGuardrail: competitor intent checker enabled (%s)",
+                competitor_intent_type,
+            )
+        except Exception as e:
+            verbose_proxy_logger.warning(
+                "ContentFilterGuardrail: failed to init competitor intent checker: %s",
+                e,
+            )
+
+    @staticmethod
+    def _normalize_patterns(
+        patterns: Optional[List[ContentFilterPattern]],
+    ) -> List[ContentFilterPattern]:
+        result: List[ContentFilterPattern] = []
+        if patterns:
+            for pattern_config in patterns:
+                if isinstance(pattern_config, dict):
+                    result.append(ContentFilterPattern(**pattern_config))
+                else:
+                    result.append(pattern_config)
+        return result
+
+    @staticmethod
+    def _normalize_blocked_words(
+        blocked_words: Optional[List[BlockedWord]],
+    ) -> List[BlockedWord]:
+        result: List[BlockedWord] = []
+        if blocked_words:
+            for word in blocked_words:
+                if isinstance(word, dict):
+                    result.append(BlockedWord(**word))
+                else:
+                    result.append(word)
+        return result
+
     @staticmethod
     def _resolve_category_file_path(file_path: str) -> str:
         """
@@ -1878,7 +1922,8 @@ class ContentFilterGuardrail(CustomGuardrail):
 
     @staticmethod
     def get_config_model():
-        from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter import \
-            LitellmContentFilterGuardrailConfigModel
+        from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter import (
+            LitellmContentFilterGuardrailConfigModel,
+        )
 
         return LitellmContentFilterGuardrailConfigModel

litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/guardrail_benchmarks/test_eval.py

@@ -69,6 +69,67 @@ def _run(checker, text: str) -> dict:
         raise
 
 
+def _print_confusion_report(label: str, metrics: dict, wrong: list) -> None:
+    """Print the confusion matrix report to stdout."""
+    print("\n")  # noqa: T201
+    print("=" * 70)  # noqa: T201
+    print(f"  {label}")  # noqa: T201
+    print("=" * 70)  # noqa: T201
+    print(f"  Total cases:  {metrics['total']}")  # noqa: T201
+    print(f"  Correct:      {metrics['tp'] + metrics['tn']}")  # noqa: T201
+    print(f"  Wrong:        {metrics['fp'] + metrics['fn']}")  # noqa: T201
+    print()  # noqa: T201
+    print(f"  TP (correctly blocked):  {metrics['tp']}")  # noqa: T201
+    print(f"  TN (correctly allowed):  {metrics['tn']}")  # noqa: T201
+    print(f"  FP (wrongly blocked):    {metrics['fp']}")  # noqa: T201
+    print(f"  FN (wrongly allowed):    {metrics['fn']}")  # noqa: T201
+    print()  # noqa: T201
+    print(f"  Precision:  {metrics['precision']:.1%}")  # noqa: T201
+    print(f"  Recall:     {metrics['recall']:.1%}")  # noqa: T201
+    print(f"  F1:         {metrics['f1']:.1%}")  # noqa: T201
+    print(f"  Accuracy:   {metrics['accuracy']:.1%}")  # noqa: T201
+    print()  # noqa: T201
+    print(f"  Latency p50:  {metrics['p50']:.1f}ms")  # noqa: T201
+    print(f"  Latency p95:  {metrics['p95']:.1f}ms")  # noqa: T201
+    print(f"  Latency avg:  {metrics['avg_lat']:.1f}ms")  # noqa: T201
+    print()  # noqa: T201
+    if wrong:
+        print("WRONG ANSWERS:")  # noqa: T201
+        for line in wrong:
+            print(line)  # noqa: T201
+    else:
+        print("ALL CASES CORRECT")  # noqa: T201
+    print("=" * 70)  # noqa: T201
+
+
+def _save_confusion_results(label: str, metrics: dict, wrong: list, rows: list) -> dict:
+    """Save confusion matrix results to a JSON file and return the result dict."""
+    os.makedirs(RESULTS_DIR, exist_ok=True)
+    safe_label = label.lower().replace(" ", "_").replace("—", "-")
+    result = {
+        "label": label,
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+        "total": metrics["total"],
+        "tp": metrics["tp"],
+        "tn": metrics["tn"],
+        "fp": metrics["fp"],
+        "fn": metrics["fn"],
+        "precision": round(metrics["precision"], 4),
+        "recall": round(metrics["recall"], 4),
+        "f1": round(metrics["f1"], 4),
+        "accuracy": round(metrics["accuracy"], 4),
+        "latency_p50_ms": round(metrics["p50"], 3),
+        "latency_p95_ms": round(metrics["p95"], 3),
+        "latency_avg_ms": round(metrics["avg_lat"], 3),
+        "wrong": wrong,
+        "rows": rows,
+    }
+    result_path = os.path.join(RESULTS_DIR, f"{safe_label}.json")
+    with open(result_path, "w") as f:
+        json.dump(result, f, indent=2)
+    return result
+
+
 def _confusion_matrix(checker, cases: List[dict], label: str):
     """Run all cases, print confusion matrix, save results JSON."""
     tp = fp = tn = fn = 0
@@ -131,62 +192,13 @@ def _confusion_matrix(checker, cases: List[dict], label: str):
     p95 = sorted_lat[int(len(sorted_lat) * 0.95)] if sorted_lat else 0
     avg_lat = sum(latencies) / len(latencies) if latencies else 0
 
-    # Print confusion matrix (noqa: T201 — intentional eval output)
-    print("\n")  # noqa: T201
-    print("=" * 70)  # noqa: T201
-    print(f"  {label}")  # noqa: T201
-    print("=" * 70)  # noqa: T201
-    print(f"  Total cases:  {total}")  # noqa: T201
-    print(f"  Correct:      {tp + tn}")  # noqa: T201
-    print(f"  Wrong:        {fp + fn}")  # noqa: T201
-    print()  # noqa: T201
-    print(f"  TP (correctly blocked):  {tp}")  # noqa: T201
-    print(f"  TN (correctly allowed):  {tn}")  # noqa: T201
-    print(f"  FP (wrongly blocked):    {fp}")  # noqa: T201
-    print(f"  FN (wrongly allowed):    {fn}")  # noqa: T201
-    print()  # noqa: T201
-    print(f"  Precision:  {precision:.1%}")  # noqa: T201
-    print(f"  Recall:     {recall:.1%}")  # noqa: T201
-    print(f"  F1:         {f1:.1%}")  # noqa: T201
-    print(f"  Accuracy:   {accuracy:.1%}")  # noqa: T201
-    print()  # noqa: T201
-    print(f"  Latency p50:  {p50:.1f}ms")  # noqa: T201
-    print(f"  Latency p95:  {p95:.1f}ms")  # noqa: T201
-    print(f"  Latency avg:  {avg_lat:.1f}ms")  # noqa: T201
-    print()  # noqa: T201
-    if wrong:
-        print("WRONG ANSWERS:")  # noqa: T201
-        for line in wrong:
-            print(line)  # noqa: T201
-    else:
-        print("ALL CASES CORRECT")  # noqa: T201
-    print("=" * 70)  # noqa: T201
-
-    # Save results
-    os.makedirs(RESULTS_DIR, exist_ok=True)
-    safe_label = label.lower().replace(" ", "_").replace("—", "-")
-    result = {
-        "label": label,
-        "timestamp": datetime.now(timezone.utc).isoformat(),
-        "total": total,
-        "tp": tp,
-        "tn": tn,
-        "fp": fp,
-        "fn": fn,
-        "precision": round(precision, 4),
-        "recall": round(recall, 4),
-        "f1": round(f1, 4),
-        "accuracy": round(accuracy, 4),
-        "latency_p50_ms": round(p50, 3),
-        "latency_p95_ms": round(p95, 3),
-        "latency_avg_ms": round(avg_lat, 3),
-        "wrong": wrong,
-        "rows": rows,
+    metrics = {
+        "total": total, "tp": tp, "tn": tn, "fp": fp, "fn": fn,
+        "precision": precision, "recall": recall, "f1": f1, "accuracy": accuracy,
+        "p50": p50, "p95": p95, "avg_lat": avg_lat,
     }
-    result_path = os.path.join(RESULTS_DIR, f"{safe_label}.json")
-    with open(result_path, "w") as f:
-        json.dump(result, f, indent=2)
-
+    _print_confusion_report(label, metrics, wrong)
+    result = _save_confusion_results(label, metrics, wrong, rows)
     return result
 
 

litellm/proxy/utils.py

@@ -14,6 +14,8 @@ from email.mime.text import MIMEText
 from typing import (
     TYPE_CHECKING,
     Any,
+    Callable,
+    Coroutine,
     Dict,
     List,
     Literal,
@@ -4009,6 +4011,19 @@ class PrismaClient:
             async with self._db_reconnect_lock:
                 return await _attempt_reconnect_inside_lock()
 
+        return await self._attempt_reconnect_with_lock_timeout(
+            _attempt_reconnect_inside_lock,
+            reason=reason,
+            lock_timeout_seconds=lock_timeout_seconds,
+        )
+
+    async def _attempt_reconnect_with_lock_timeout(
+        self,
+        reconnect_fn: Callable[[], Coroutine[Any, Any, bool]],
+        reason: str,
+        lock_timeout_seconds: float,
+    ) -> bool:
+        """Acquire the reconnect lock with a timeout, then run reconnect_fn."""
         lock_acquired_by_timeout_task = False
 
         async def _acquire_reconnect_lock() -> bool:
@@ -4056,7 +4071,7 @@ class PrismaClient:
             return False
 
         try:
-            return await _attempt_reconnect_inside_lock()
+            return await reconnect_fn()
         finally:
             self._db_reconnect_lock.release()