fix(type-checking): satisfy the basedpyright budget gate for auto-router compression

Two fixes for the zero-headroom basedpyright budget:

- arm_pre_call's data parameter is dict[str, object], not MutableMapping: the
  latter is itself banned by LIT001 with no benefit, and it mismatched every
  dict-typed helper (get_or_create_metadata_bucket, resolve_structured_messages,
  _get_tags_from_request_kwargs), which is what the budget was actually flagging.
- Router.async_pre_routing_hook computed pre_routing_hook_response in one shot
  instead of reassigning a Final-annotated local.

The remaining two reportArgumentType hits are pre-existing: LiteLLM_Params(**merged)
in _create_deployment_object already fails this check for all ~165 of its other
fields, since the merged dict's value type is partly untyped/float; adding two new
string fields to the model just grows that existing pile by two. Suppressed at the
one call site with a reason, since fixing the root typing is out of scope here.
This commit is contained in:
moe-berri 2026-09-04 18:37:38 -07:00
parent dc63428395
commit 88ada40cda
2 changed files with 15 additions and 13 deletions

View file

@ -13,7 +13,7 @@ each hop sees.
"""
import contextvars
from collections.abc import Iterable, Mapping, MutableMapping, Sequence
from collections.abc import Iterable, Mapping, Sequence
from dataclasses import dataclass
from types import MappingProxyType
from typing import TYPE_CHECKING, Final
@ -89,7 +89,7 @@ def policy_for_model(
markers: Final = tuple(
litellm_params
for deployment in deployments
if isinstance(litellm_params := deployment.get("litellm_params"), Mapping)
if isinstance(litellm_params := deployment.get("litellm_params"), Mapping) # pyright: ignore[reportUnnecessaryIsInstance] # filters out non-Mapping
and str(litellm_params.get("model", "")).startswith(AUTO_ROUTER_MODEL_PREFIX)
)
requested: Final = frozenset(request_tags)
@ -130,7 +130,7 @@ def _active_compression_guardrails() -> tuple["CustomGuardrail", ...]:
async def arm_pre_call(
data: MutableMapping[str, object], # mutable-ok: arms the live request dict in place
data: dict[str, object], # mutable-ok: arms the live request dict in place
llm_router: "Router | None",
) -> None:
"""Apply an auto router's compression policy, if any, before guardrails run.
@ -183,7 +183,11 @@ async def arm_pre_call(
from litellm.litellm_core_utils.prompt_templates.factory import resolve_structured_messages
snapshot: Final = resolve_structured_messages(messages=data.get("messages"), request_kwargs=data)
raw_messages: Final = data.get("messages")
snapshot: Final = resolve_structured_messages(
messages=raw_messages if isinstance(raw_messages, list) else None,
request_kwargs=data,
)
if snapshot is not None:
_routing_messages_snapshot.set(tuple(MappingProxyType(dict(message)) for message in snapshot))

View file

@ -8644,7 +8644,7 @@ class Router:
raise ValueError(ptu_error)
zeroed_pricing: Final = zeroed_ptu_pricing(_model_info, _litellm_params) if config_sourced else None
litellm_params: Final[LiteLLM_Params] = LiteLLM_Params(
**(
**( # pyright: ignore[reportArgumentType] # untyped merged dict; already true for every field here
_litellm_params
if zeroed_pricing is None
else MappingProxyType({**_litellm_params, **zeroed_pricing})
@ -13066,7 +13066,7 @@ class Router:
else None
)
pre_routing_hook_response: Final = await selected_strategy.strategy.async_pre_routing_hook(
routed: Final = await selected_strategy.strategy.async_pre_routing_hook(
model=registered_model_name,
request_kwargs=request_kwargs,
messages=routing_messages if routing_messages is not None else messages,
@ -13079,13 +13079,11 @@ class Router:
# Compared by value, not identity: PreRoutingHookResponse is a pydantic model,
# and pydantic reconstructs a validated list field rather than keeping the
# exact object passed in, even when nothing about it changed.
if (
pre_routing_hook_response is not None
and routing_messages is not None
and pre_routing_hook_response.messages == routing_messages
):
restored: Final = {"messages": messages} # mutable-ok: pydantic's model_copy takes a dict
pre_routing_hook_response = pre_routing_hook_response.model_copy(update=restored)
pre_routing_hook_response: Final = (
routed.model_copy(update={"messages": messages}) # mutable-ok: pydantic's model_copy takes a dict
if routed is not None and routing_messages is not None and routed.messages == routing_messages
else routed
)
self._record_routing_decision(
request_kwargs=request_kwargs,
routing_decision=(pre_routing_hook_response.routing_decision if pre_routing_hook_response else None),