scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-09 03:18:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 108

fix: place lint suppressions on the flagged annotation lines

Browse source
This commit is contained in:
Devin AI 2026-09-23 02:36:27 +00:00 • committed by mateo
parent 8e4c26fea3
commit 46f81ba3e0
10 changed files with 93 additions and 47 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
litellm/litellm_core_utils/internal_call_metadata.py
View file

@ -165,6 +165,6 @@ def parent_session_kwargs(request_kwargs: Mapping[str, object] | None) -> Mappin
def effective_turn_off_message_logging(request_kwargs: Mapping[str, object] | None) -> bool | None:
return initialize_standard_callback_dynamic_params(dict(request_kwargs) if request_kwargs else None).get(
"turn_off_message_logging"
)
return initialize_standard_callback_dynamic_params(
dict(request_kwargs) if request_kwargs else None # mutable-ok: callback params take a mutable dict copy
).get("turn_off_message_logging")

4
litellm/proxy/auth/auth_checks.py
View file

@ -5116,7 +5116,9 @@ async def _check_team_member_model_access(
):
return # no per-member restriction — inherit team-level check
member_allowed_models: Final[list[str]] = loaded_membership.litellm_budget_table.allowed_models
member_allowed_models: Final[list[str]] = ( # mutable-ok: allowed_models is a prisma model list consumed read-only
loaded_membership.litellm_budget_table.allowed_models
)
try:
_can_object_call_model(
model=model,

2
litellm/proxy/auth/team_grants.py
View file

@ -59,7 +59,7 @@ class TeamGrants(TypedDict, total=False):
team_tpd_limit: ReadOnly[int | None]
team_max_budget: ReadOnly[float | None]
team_soft_budget: ReadOnly[float | None]
team_model_max_budget: ReadOnly[dict[str, object] | None]
team_model_max_budget: ReadOnly[dict[str, object] | None] # mutable-ok: prisma table field typed loosely
team_spend: ReadOnly[float | None]
team_models: ReadOnly[Sequence[str]]
team_blocked: ReadOnly[bool]

4
litellm/proxy/guardrails/guardrail_hooks/typesafe/__init__.py
View file

@ -20,8 +20,8 @@ if TYPE_CHECKING:
def _coerce_event_hook(
mode: str | list[str] | Mode,
) -> GuardrailEventHooks | list[GuardrailEventHooks] | Mode:
mode: str | list[str] | Mode, # mutable-ok: event hook unions accept an ordered list
) -> GuardrailEventHooks | list[GuardrailEventHooks] | Mode: # mutable-ok: event hook unions accept an ordered list
if isinstance(mode, Mode):
return mode
if isinstance(mode, list):

47
litellm/proxy/guardrails/guardrail_hooks/typesafe/typesafe.py
View file

@ -60,14 +60,14 @@ _STR_OBJECT_DICT_ADAPTER: Final = TypeAdapter(dict[str, object])
_OBJECT_LIST_ADAPTER: Final = TypeAdapter(list[object])
def _as_str_object_dict(value: object) -> dict[str, object] | None:
def _as_str_object_dict(value: object) -> dict[str, object] | None: # mutable-ok: parsed JSON payload is dict-shaped
try:
return _STR_OBJECT_DICT_ADAPTER.validate_python(value)
except ValidationError:
return None
def _as_object_list(value: object) -> list[object] | None:
def _as_object_list(value: object) -> list[object] | None: # mutable-ok: parsed JSON payload is list-shaped
try:
return _OBJECT_LIST_ADAPTER.validate_python(value)
except ValidationError:
@ -120,7 +120,9 @@ def _question_instructions(question_id: str) -> str:
)
def _tool_call_entry(tool_call: object) -> dict[str, object] | None:
def _tool_call_entry(
tool_call: object,
) -> dict[str, object] | None: # mutable-ok: tool call entries are request-payload dicts
parsed_call = _as_str_object_dict(tool_call)
if parsed_call is None:
return None
@ -129,7 +131,9 @@ def _tool_call_entry(tool_call: object) -> dict[str, object] | None:
return {"name": fn.get("name"), "arguments": fn.get("arguments")} # mutable-ok: serialized to JSON
def _tool_call_entries(assistant_message: Mapping[str, object]) -> tuple[dict[str, object], ...]:
def _tool_call_entries(
assistant_message: Mapping[str, object],
) -> tuple[dict[str, object], ...]: # mutable-ok: tool call entries are request-payload dicts
tool_calls: Final = _as_object_list(assistant_message.get("tool_calls"))
if tool_calls is None:
return ()
@ -158,7 +162,10 @@ class TypeSafeGuardrail(CustomGuardrail):
max_result_chars_in_state: int | None = None,
unreachable_fallback: str | None = None,
guardrail_name: str | None = None,
event_hook: GuardrailEventHooks | list[GuardrailEventHooks] | Mode | None = None,
event_hook: GuardrailEventHooks # mutable-ok: event hook unions accept an ordered list
| list[GuardrailEventHooks]
| Mode
| None = None,
default_on: bool = False,
async_handler: AsyncHTTPHandler | None = None,
) -> None:
@ -190,7 +197,11 @@ class TypeSafeGuardrail(CustomGuardrail):
default_on=default_on,
)
def _handle_failure(self, error: str, log_detail: dict[str, object]) -> None:
def _handle_failure(
self,
error: str,
log_detail: dict[str, object], # mutable-ok: log detail record is dict-shaped
) -> None:
"""fail_open logs and returns; fail_closed raises a generic 502 (upstream bodies stay in server logs)."""
if self.unreachable_fallback == "fail_open":
verbose_proxy_logger.warning(
@ -202,7 +213,10 @@ class TypeSafeGuardrail(CustomGuardrail):
verbose_proxy_logger.error("TypeSafe: %s. detail=%s", error, log_detail)
raise HTTPException(status_code=502, detail={"error": error}) # mutable-ok: FastAPI wants a dict detail
def _candidate_exchanges(self, messages: Sequence[dict[str, object]]) -> tuple[tuple[int, ...], ...]:
def _candidate_exchanges(
self,
messages: Sequence[dict[str, object]], # mutable-ok: message dicts come from the request payload
) -> tuple[tuple[int, ...], ...]:
"""Completed tool exchanges eligible for evaluation: unprotected, and long enough to be worth a call."""
protected: Final = _protected_indices(messages)
candidates: Final = tuple(
@ -216,7 +230,10 @@ class TypeSafeGuardrail(CustomGuardrail):
return candidates[-_MAX_EXCHANGES_EVALUATED:]
@staticmethod
def _exchange_tool_text(messages: Sequence[dict[str, object]], group: tuple[int, ...]) -> str:
def _exchange_tool_text(
messages: Sequence[dict[str, object]],
group: tuple[int, ...], # mutable-ok: message dicts come from the request payload
) -> str:
return "".join(
content_to_text(messages[index].get("content"))
for index in group[1:]
@ -224,7 +241,9 @@ class TypeSafeGuardrail(CustomGuardrail):
)
def _build_state(
self, messages: Sequence[dict[str, object]], candidates: tuple[tuple[int, ...], ...]
self,
messages: Sequence[dict[str, object]], # mutable-ok: candidate groups index request message dicts
candidates: tuple[tuple[int, ...], ...],
) -> dict[str, object]:
task: Final = next(
(
@ -249,7 +268,9 @@ class TypeSafeGuardrail(CustomGuardrail):
return {"task": task, "system": system, "tool_exchanges": tool_exchanges} # mutable-ok: serialized to JSON
async def _call_systemone(
self, state: dict[str, object], question_ids: Sequence[str]
self,
state: dict[str, object], # mutable-ok: state dict is the parsed log record
question_ids: Sequence[str],
) -> _JevSystemOneResponse | None:
"""Returns the response, or None when the service failed and fail_open applies."""
payload: Final[dict[str, object]] = { # mutable-ok: serialized to JSON by httpx
@ -275,8 +296,8 @@ class TypeSafeGuardrail(CustomGuardrail):
)
except asyncio.CancelledError:
raise
except Exception as e:
detail: Final[dict[str, object]] = (
except Exception as e: # noqa: BLE001 # fail-open guardrail must not leak provider exceptions
detail: Final[dict[str, object]] = ( # mutable-ok: log detail record is dict-shaped
{ # mutable-ok: log detail record
"error_type": type(e).__name__,
"detail": str(e),
@ -318,7 +339,7 @@ class TypeSafeGuardrail(CustomGuardrail):
async def apply_guardrail(
self,
inputs: GenericGuardrailAPIInputs,
request_data: dict[str, object],
request_data: dict[str, object], # mutable-ok: request data is dict-shaped
input_type: Literal["request", "response"],
logging_obj: LiteLLMLoggingObj | None = None,
) -> GenericGuardrailAPIInputs:

6
litellm/proxy/management_endpoints/model_management_endpoints.py
View file

@ -2029,8 +2029,10 @@ async def update_model(
)
### MERGE WITH EXISTING DATA ###
_mp: Final[dict[str, object]] = model_params.litellm_params.dict()
merged_dictionary: Final = {
_mp: Final[dict[str, object]] = ( # mutable-ok: litellm_params dict() output is the merge source
model_params.litellm_params.dict()
)
merged_dictionary: Final = { # mutable-ok: merged params dict feeds litellm_params
key: _existing_litellm_params_dict[key] if value is None else encrypted_params[key]
for key, value in _mp.items()
if value is not None or _existing_litellm_params_dict.get(key) is not None

4
litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py
View file

@ -474,7 +474,7 @@ async def mistral_proxy_route(
methods=["GET", "POST", "PUT", "DELETE", "PATCH"], # mutable-ok: FastAPI route metadata requires a list
tags=["TypeSafe AI Pass-through", "pass-through"], # mutable-ok: FastAPI route metadata requires a list
)
async def typesafe_proxy_route(
async def typesafe_proxy_route( # noqa: ANN201 # FastAPI route returns the endpoint_func response object
endpoint: str,
request: Request,
fastapi_response: Response,
@ -510,7 +510,7 @@ async def typesafe_proxy_route(
methods=["GET", "POST", "PUT", "DELETE", "PATCH"], # mutable-ok: FastAPI route metadata requires a list
tags=["OpenRouter Pass-through", "pass-through"], # mutable-ok: FastAPI route metadata requires a list
)
async def openrouter_proxy_route(
async def openrouter_proxy_route( # noqa: ANN201 # FastAPI route returns the endpoint_func response object
endpoint: str,
request: Request,
fastapi_response: Response,

2
litellm/proxy/pass_through_endpoints/llm_provider_handlers/typesafe_passthrough_logging_handler.py
View file

@ -66,7 +66,7 @@ class TypeSafePassthroughLoggingHandler:
cache_hit: bool,
request_body: Mapping[str, object],
custom_llm_provider: str,
**kwargs: object,
**kwargs: object, # kwargs-ok: logging handler forwards the SDK kwargs contract
) -> PassThroughEndpointLoggingTypedDict:
response: Final = _parse_typesafe_response(response_body)
response_model: Final = response.model

15
litellm/router_strategy/complexity_router/complexity_router.py
View file

@ -408,7 +408,7 @@ def _human_text(content: object, marker_pairs: tuple[tuple[str, str], ...] = _DE
def _encrypted_classifier_task(
request_kwargs: Mapping[str, object] | None,
marker_pairs: tuple[tuple[str, str], ...],
) -> dict[str, object] | None:
) -> dict[str, object] | None: # mutable-ok: request payload is dict-shaped
from litellm.litellm_core_utils.prompt_templates.factory import resolve_structured_messages
raw_input: Final = (request_kwargs or EMPTY_MAPPING).get("input")
@ -422,7 +422,12 @@ def _encrypted_classifier_task(
(
item
for item in reversed(items)
if (messages := resolve_structured_messages(messages=None, request_kwargs={"input": [item]}))
if (
messages := resolve_structured_messages(
messages=None,
request_kwargs={"input": [item]}, # mutable-ok: request_kwargs wire shape is a plain dict
)
)
and any(_iter_human_asks_newest_first(messages, marker_pairs))
),
None,
@ -435,9 +440,11 @@ def _encrypted_classifier_task(
return None
if not any(part.get("type") == "encrypted_content" and part.get("encrypted_content") for part in parts):
return None
return {
return { # mutable-ok: wire body is a plain dict
**current,
"content": [part for part in parts if part.get("type") in ("input_text", "encrypted_content")],
"content": [ # mutable-ok: content blocks are a plain list
part for part in parts if part.get("type") in ("input_text", "encrypted_content")
],
}

50
tests/test_litellm/proxy/management_endpoints/test_model_management_endpoints.py
View file

@ -1198,7 +1198,9 @@ class TestTeamModelUpdate:
patch(
"litellm.proxy.management_endpoints.model_management_endpoints.team_model_add"
) as mock_team_model_add,
patch("litellm.proxy.management_endpoints.model_management_endpoints.update_team") as mock_update_team,
patch(
"litellm.proxy.management_endpoints.model_management_endpoints.update_team"
) as mock_update_team, # test-quality-ok: [TQ008] the endpoint reads this collaborator from its import site, so patching internals is the only injection point
):
result = await _update_team_model_in_db(
db_model=db_model,
@ -1252,8 +1254,12 @@ class TestTeamModelUpdate:
)
with (
patch("litellm.proxy.management_endpoints.model_management_endpoints.team_model_delete") as mock_delete,
patch("litellm.proxy.management_endpoints.model_management_endpoints.team_model_add") as mock_add,
patch(
"litellm.proxy.management_endpoints.model_management_endpoints.team_model_delete"
) as mock_delete, # test-quality-ok: [TQ008] the endpoint reads this collaborator from its import site, so patching internals is the only injection point
patch(
"litellm.proxy.management_endpoints.model_management_endpoints.team_model_add"
) as mock_add, # test-quality-ok: [TQ008] the endpoint reads this collaborator from its import site, so patching internals is the only injection point
):
await _update_existing_team_model_assignment(
team_id="team_123",
@ -1294,8 +1300,12 @@ class TestTeamModelUpdate:
)
with (
patch("litellm.proxy.management_endpoints.model_management_endpoints.team_model_delete") as mock_delete,
patch("litellm.proxy.management_endpoints.model_management_endpoints.team_model_add") as mock_add,
patch(
"litellm.proxy.management_endpoints.model_management_endpoints.team_model_delete"
) as mock_delete, # test-quality-ok: [TQ008] the endpoint reads this collaborator from its import site, so patching internals is the only injection point
patch(
"litellm.proxy.management_endpoints.model_management_endpoints.team_model_add"
) as mock_add, # test-quality-ok: [TQ008] the endpoint reads this collaborator from its import site, so patching internals is the only injection point
):
await _update_existing_team_model_assignment(
team_id="team_123",
@ -1375,8 +1385,12 @@ class TestTeamModelUpdate:
)
with (
patch("litellm.proxy.management_endpoints.model_management_endpoints.team_model_delete") as mock_delete,
patch("litellm.proxy.management_endpoints.model_management_endpoints.team_model_add") as mock_add,
patch(
"litellm.proxy.management_endpoints.model_management_endpoints.team_model_delete"
) as mock_delete, # test-quality-ok: [TQ008] the endpoint reads this collaborator from its import site, so patching internals is the only injection point
patch(
"litellm.proxy.management_endpoints.model_management_endpoints.team_model_add"
) as mock_add, # test-quality-ok: [TQ008] the endpoint reads this collaborator from its import site, so patching internals is the only injection point
):
await _update_existing_team_model_assignment(
team_id="team_123",
@ -3175,16 +3189,16 @@ class TestPatchModelRowDeletedBeforeWrite:
mock_prisma.db.litellm_proxymodeltable.update = AsyncMock(return_value=None)
with (
patch(
patch( # test-quality-ok: [TQ008] the endpoint reads this collaborator from its import site, so patching internals is the only injection point
"litellm.proxy.proxy_server.prisma_client", mock_prisma
), # test-quality-ok: proxy_server module global is the endpoint's only injection point
patch(
patch( # test-quality-ok: [TQ008] the endpoint reads this collaborator from its import site, so patching internals is the only injection point
"litellm.proxy.proxy_server.llm_router", MagicMock(**{"get_model_ids.return_value": ["m1"]})
), # test-quality-ok: proxy_server module global is the endpoint's only injection point
patch(
patch( # test-quality-ok: [TQ008] the endpoint reads this collaborator from its import site, so patching internals is the only injection point
"litellm.proxy.proxy_server.store_model_in_db", True
), # test-quality-ok: proxy_server module global is the endpoint's only injection point
patch(
patch( # test-quality-ok: [TQ008] the endpoint reads this collaborator from its import site, so patching internals is the only injection point
"litellm.proxy.proxy_server.premium_user", True
), # test-quality-ok: proxy_server module global is the endpoint's only injection point
patch( # test-quality-ok: stubs the auth gate so the test exercises the not-found branch under test
@ -4311,25 +4325,25 @@ class TestTeamMemberAutoRouterWrites:
@contextlib.contextmanager
def _environment(self, database: MagicMock, row: LiteLLM_ProxyModelTable) -> Iterator[None]:
with (
patch(
patch( # test-quality-ok: [TQ008] the endpoint reads this collaborator from its import site, so patching internals is the only injection point
"litellm.proxy.proxy_server.prisma_client", database
), # test-quality-ok: [TQ008] endpoint storage singleton injection
patch(
patch( # test-quality-ok: [TQ008] the endpoint reads this collaborator from its import site, so patching internals is the only injection point
"litellm.proxy.proxy_server.llm_router", self._catalog()
), # test-quality-ok: [TQ008] inject real destination model catalog
patch(
patch( # test-quality-ok: [TQ008] the endpoint reads this collaborator from its import site, so patching internals is the only injection point
"litellm.proxy.proxy_server.store_model_in_db", True
), # test-quality-ok: [TQ008] endpoint storage mode singleton
patch(
patch( # test-quality-ok: [TQ008] the endpoint reads this collaborator from its import site, so patching internals is the only injection point
"litellm.proxy.proxy_server.premium_user", True
), # test-quality-ok: [TQ008] inject licensed process state
patch(
patch( # test-quality-ok: [TQ008] the endpoint reads this collaborator from its import site, so patching internals is the only injection point
"litellm.proxy.management_endpoints.model_management_endpoints.publish_config_change", new=AsyncMock()
), # test-quality-ok: [TQ008] pubsub I/O boundary
patch(
patch( # test-quality-ok: [TQ008] the endpoint reads this collaborator from its import site, so patching internals is the only injection point
"litellm.proxy.management_endpoints.model_management_endpoints.create_object_audit_log", new=AsyncMock()
), # test-quality-ok: [TQ008] audit database I/O boundary
patch(
patch( # test-quality-ok: [TQ008] the endpoint reads this collaborator from its import site, so patching internals is the only injection point
"litellm.proxy.management_endpoints.model_management_endpoints.clear_cache",
new=AsyncMock(
return_value=ReconcileOutcome( # test-quality-ok: [TQ008] model reload I/O boundary