scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-13 03:59:12 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 103

fix(proxy): replace Any annotations with starlette.types in guard middleware

Browse source 
Keeps the strict-ruff ANN401 delta at zero.
This commit is contained in:
Renn F 2026-10-11 01:39:40 +02:00
parent 1605b43a1d
commit c27bffb33e
No known key found for this signature in database
GPG key ID: B48363D3A3A4CF78
1 changed files with 8 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
litellm/proxy/middleware/guard_middleware.py
View file

@ -30,7 +30,9 @@ Design constraints honored here:
from __future__ import annotations
import os
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING
from starlette.types import ASGIApp, Receive, Scope, Send
from litellm._logging import verbose_proxy_logger
@ -74,7 +76,7 @@ def _build_guard_config() -> SecurityConfig:
"""Build the SecurityConfig from the LITELLM_GUARD_* environment variables."""
from guard import SecurityConfig
kwargs: dict[str, Any] = {
kwargs: dict[str, object] = {
"enable_rate_limiting": True,
"rate_limit": _env_int("LITELLM_GUARD_RATE_LIMIT", 100),
"rate_limit_window": _env_int("LITELLM_GUARD_RATE_LIMIT_WINDOW", 60),
@ -124,12 +126,12 @@ class GuardMiddleware:
admits flows through guard's pipeline and then the normal proxy stack.
"""
def __init__(self, app: Any) -> None:
def __init__(self, app: ASGIApp) -> None:
self.app = app
self._guard_asgi: Any = None
self._guard_asgi: ASGIApp | None = None
self._guard_failed = False
def _ensure_guard(self, scope: Any) -> bool:
def _ensure_guard(self, scope: Scope) -> bool:
if self._guard_asgi is not None or self._guard_failed:
return not self._guard_failed
try:
@ -148,7 +150,7 @@ class GuardMiddleware:
)
return False
async def __call__(self, scope: Any, receive: Any, send: Any) -> None:
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
if scope["type"] != "http" or not guard_enabled():
await self.app(scope, receive, send)
return