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

chore: type-discipline cleanup for canonical model resolution

Browse source 
basedpyright: new module at 0 errors; modified files back to baseline
(+1 structural 'Return type is Any' on the new route return, matching the
file's existing pattern on every route return). ruff clean.

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
abhi 2026-08-15 08:36:36 -07:00
parent 55c499100e
commit d952e5e87e
3 changed files with 35 additions and 22 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
litellm/proxy/route_llm_request.py
View file

@ -670,9 +670,10 @@ async def route_request(
# identity-attested only, and runs here -- after every configured route,
# including wildcards and default_deployment, has declined -- so it can only
# turn a hard failure into a success, never re-point working traffic.
if llm_router is not None and isinstance(data.get("model"), str):
requested_model: Final[object] = data.get("model") # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] - data is an untyped request dict
if llm_router is not None and isinstance(requested_model, str):
canonical_target: Final = llm_router.resolve_canonical_model_name(
model=data["model"],
model=requested_model,
request_team_id=team_id,
)
if canonical_target is not None:
@ -686,7 +687,9 @@ async def route_request(
# the same 400 it gets today, revealing nothing about the target.
target_allowed = True
if user_api_key_dict is not None:
from litellm.proxy.auth.auth_checks import can_key_call_model
from litellm.proxy.auth.auth_checks import (
can_key_call_model, # pyright: ignore[reportUnknownVariableType] - auth_checks is partially typed
)
try:
await can_key_call_model(
@ -701,11 +704,11 @@ async def route_request(
# Preserve the client's spelling for spend logs / debugging --
# after the rewrite it is otherwise invisible downstream.
metadata_field: Final = "litellm_metadata" if "litellm_metadata" in data else "metadata"
existing_metadata = data.get(metadata_field)
existing_metadata: object = data.get(metadata_field) # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] - data is an untyped request dict
if isinstance(existing_metadata, dict):
existing_metadata.setdefault("requested_model", data["model"])
existing_metadata.setdefault("requested_model", requested_model) # pyright: ignore[reportUnknownMemberType] - metadata dict is untyped
else:
data[metadata_field] = {"requested_model": data["model"]}
data[metadata_field] = {"requested_model": requested_model}
data["model"] = canonical_target
return getattr(llm_router, f"{route_type}")(**data)

14
litellm/router.py
View file

@ -104,17 +104,17 @@ from litellm.router_utils.batch_utils import (
replace_model_in_jsonl,
should_replace_model_in_jsonl,
)
from litellm.router_utils.client_initalization_utils import InitalizeCachedClient
from litellm.router_utils.clientside_credential_handler import (
get_dynamic_litellm_params,
is_clientside_credential,
)
from litellm.router_utils.canonical_model_resolution import (
build_canonical_index,
)
from litellm.router_utils.canonical_model_resolution import (
lookup as canonical_lookup,
)
from litellm.router_utils.client_initalization_utils import InitalizeCachedClient
from litellm.router_utils.clientside_credential_handler import (
get_dynamic_litellm_params,
is_clientside_credential,
)
from litellm.router_utils.common_utils import (
_is_proxy_admin_request,
filter_team_based_models,
@ -10716,7 +10716,9 @@ class Router:
cost_generation: Final = get_model_cost_mutation_generation()
if self._canonical_model_index is None or self._canonical_model_index_cost_generation != cost_generation:
try:
self._canonical_model_index = build_canonical_index(self.model_list)
self._canonical_model_index = build_canonical_index(
cast("list[DeploymentTypedDict]", self.model_list)
)
except Exception as exc:
# Never let index construction brick a router: degrade to
# 'strict' behaviour instead.

28
litellm/router_utils/canonical_model_resolution.py
View file

@ -1,3 +1,8 @@
# pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportAny=false
# litellm.model_cost is loaded dynamically from a JSON price map (see
# litellm.litellm_core_utils.get_model_cost_map) and is untyped at that
# boundary, same as every other reader of it in litellm/utils.py. Every value
# pulled from it here is re-validated with isinstance before use.
"""Same-provider canonical model-name resolution.
Harnesses hardcode concrete model IDs. A client that asks for
@ -31,13 +36,11 @@ See also ``Router.resolve_canonical_model_name``.
"""
from collections.abc import Mapping
from typing import TYPE_CHECKING, Final
from typing import Final
import litellm
from litellm._logging import verbose_router_logger
if TYPE_CHECKING:
from litellm.types.router import DeploymentTypedDict
from litellm.types.router import DeploymentTypedDict
# Cost-map fields that must match exactly for two names to be called the same
# model. Pricing equality is a tripwire against false identity, not the
@ -59,7 +62,7 @@ _AMBIGUOUS: Final = None
def _cost_map_entry(model: str) -> Mapping[str, object] | None:
"""The cost-map entry for ``model``, or None when absent."""
entry: Final = litellm.model_cost.get(model)
entry = litellm.model_cost.get(model)
return entry if isinstance(entry, Mapping) else None
@ -135,7 +138,7 @@ def _undated_variants(model: str) -> tuple[str, ...]:
def build_canonical_index(
deployments: list["DeploymentTypedDict"],
deployments: list[DeploymentTypedDict],
) -> dict[tuple[str, str], str | None]:
"""Map ``(provider, canonical_name) -> model group`` for ``deployments``.
@ -151,10 +154,15 @@ def build_canonical_index(
for deployment in deployments:
try:
model_group = deployment.get("model_name")
# ``model_name``/``model`` are typed Required[str], but this index is
# built from operator config and DB rows that can violate the type,
# so both are validated at runtime rather than trusted.
model_group: object = deployment.get("model_name")
litellm_params = deployment.get("litellm_params") or {}
underlying = litellm_params.get("model") if isinstance(litellm_params, Mapping) else None
if not isinstance(model_group, str) or not isinstance(underlying, str):
underlying: object = litellm_params.get("model")
if not isinstance(model_group, str) or not isinstance( # pyright: ignore[reportUnnecessaryIsInstance] - config/DB rows can violate the TypedDict
underlying, str
):
continue
identity = canonicalize(underlying)
@ -179,7 +187,7 @@ def build_canonical_index(
if _same_model_per_cost_map(canonical_name, candidate):
spellings.append(candidate)
for dated, entry in litellm.model_cost.items():
if not isinstance(entry, Mapping) or dated in spellings:
if not isinstance(dated, str) or not isinstance(entry, Mapping) or dated in spellings:
continue
if _undated_variants(dated) == (canonical_name,) and _same_model_per_cost_map(canonical_name, dated):
spellings.append(dated)