scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-05 02:41:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 49

fix(ci): modernize type hints to clear ruff strict-rule budget

Browse source
This commit is contained in:
Krrish Dholakia 2026-07-14 21:12:57 -07:00
parent 99a72131b8
commit 0ab46cfd5d
6 changed files with 61 additions and 69 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
litellm/proxy/client/cli/commands/autoroute/commands.py
View file

@ -4,15 +4,13 @@ import secrets
import signal
import threading
from types import FrameType
from typing import Dict, Optional
import click
import yaml
from pydantic import JsonValue, TypeAdapter
from ..up import CLAUDE_SETTINGS_PATH
from ..up import CLAUDE_SETTINGS_PATH, load_json_or_empty, restore_claude_settings, write_backup
from ..up import BackupRecord as ClaudeBackupRecord
from ..up import load_json_or_empty, restore_claude_settings, write_backup
from .process import (
AUTOROUTE_DIR,
CONFIG_PATH,
@ -34,7 +32,7 @@ from .wizard import run_configure_wizard
AUTOROUTE_BACKUP_PATH = AUTOROUTE_DIR / "claude_settings_backup.json"
_GENERATED_CONFIG_ADAPTER = TypeAdapter(Dict[str, JsonValue])
_GENERATED_CONFIG_ADAPTER = TypeAdapter(dict[str, JsonValue])
def _mint_and_embed_master_key() -> str:
@ -49,11 +47,11 @@ def _mint_and_embed_master_key() -> str:
with open(CONFIG_PATH, "r") as f:
generated = _GENERATED_CONFIG_ADAPTER.validate_python(yaml.safe_load(f))
general_settings = generated.get("general_settings")
updated_settings: Dict[str, JsonValue] = {
updated_settings: dict[str, JsonValue] = {
**(general_settings if isinstance(general_settings, dict) else {}),
"master_key": master_key,
}
updated: Dict[str, JsonValue] = {**generated, "general_settings": updated_settings}
updated: dict[str, JsonValue] = {**generated, "general_settings": updated_settings}
with open(CONFIG_PATH, "w") as f:
yaml.safe_dump(updated, f, sort_keys=False)
CONFIG_PATH.chmod(0o600)
@ -122,7 +120,7 @@ def up() -> None:
restore_claude_settings(CLAUDE_SETTINGS_PATH, AUTOROUTE_BACKUP_PATH)
click.echo("\nStopped ephemeral proxy and restored Claude Code settings.")
def _handle_signal(_signum: int, _frame: Optional[FrameType]) -> None:
def _handle_signal(_signum: int, _frame: FrameType | None) -> None:
stop_event.set()
signal.signal(signal.SIGINT, _handle_signal)
@ -139,7 +137,7 @@ def up() -> None:
@autoroute_group.command("down")
def down() -> None:
"""Restore Claude Code settings and stop a leftover ephemeral proxy, if any"""
record: Optional[PidRecord] = read_pid_record()
record: PidRecord | None = read_pid_record()
if record is not None and is_running(record.pid):
terminate(record.pid)
click.echo(f"Stopped leftover ephemeral proxy (pid {record.pid}).")

46
litellm/proxy/client/cli/commands/autoroute/config.py
View file

@ -1,8 +1,8 @@
from typing import Dict, FrozenSet, List, Literal, Tuple, Union
from typing import Literal, Union
from pydantic import BaseModel, ConfigDict, Field, JsonValue, TypeAdapter
TIER_NAMES: Tuple[str, ...] = ("SIMPLE", "MEDIUM", "COMPLEX", "REASONING")
TIER_NAMES: tuple[str, ...] = ("SIMPLE", "MEDIUM", "COMPLEX", "REASONING")
AUTOROUTER_MODEL_NAME = "autorouter"
@ -32,10 +32,10 @@ class _RawModelGroup(BaseModel):
output_cost_per_token: float | None = None
_RAW_MODEL_GROUPS_ADAPTER = TypeAdapter(List[_RawModelGroup])
_RAW_MODEL_GROUPS_ADAPTER = TypeAdapter(list[_RawModelGroup])
def parse_discovered_models(raw: List[JsonValue]) -> Tuple[DiscoveredModel, ...]:
def parse_discovered_models(raw: list[JsonValue]) -> tuple[DiscoveredModel, ...]:
"""Validate a raw `/model_group/info` response into typed models."""
parsed = _RAW_MODEL_GROUPS_ADAPTER.validate_python(raw)
return tuple(
@ -52,11 +52,11 @@ def parse_discovered_models(raw: List[JsonValue]) -> Tuple[DiscoveredModel, ...]
)
def chat_models(models: Tuple[DiscoveredModel, ...]) -> Tuple[DiscoveredModel, ...]:
def chat_models(models: tuple[DiscoveredModel, ...]) -> tuple[DiscoveredModel, ...]:
return tuple(m for m in models if m.mode == "chat")
def embedding_models(models: Tuple[DiscoveredModel, ...]) -> Tuple[DiscoveredModel, ...]:
def embedding_models(models: tuple[DiscoveredModel, ...]) -> tuple[DiscoveredModel, ...]:
return tuple(m for m in models if m.mode == "embedding")
@ -91,7 +91,7 @@ SemanticMatchingChoice = Union[NoSemanticMatching, SemanticMatching]
# Satisfies complexity_router's "semantic matching requires non-empty keyword_tier_rules"
# invariant with a sane starting point; the generated config.yaml can be hand-edited afterward.
_DEFAULT_KEYWORD_TIER_RULES: Tuple[Dict[str, JsonValue], ...] = (
_DEFAULT_KEYWORD_TIER_RULES: tuple[dict[str, JsonValue], ...] = (
{"keywords": ["hi", "hello", "thanks"], "tier": "SIMPLE"},
{"keywords": ["explain", "how does"], "tier": "MEDIUM"},
{"keywords": ["refactor", "implement", "debug"], "tier": "COMPLEX"},
@ -106,17 +106,17 @@ class AutorouteConfig(BaseModel):
api_key: str
# Each tier maps to a pool of one or more models; complexity_router picks randomly among
# them per request (or, in adaptive mode, learns which to prefer within the pool).
tiers: Dict[str, Tuple[str, ...]]
tiers: dict[str, tuple[str, ...]]
default_model: str
classifier: ClassifierChoice = Field(default_factory=HeuristicClassifier)
semantic_matching: SemanticMatchingChoice = Field(default_factory=NoSemanticMatching)
adaptive: bool = False
def validate_config(config: AutorouteConfig, discovered: Tuple[DiscoveredModel, ...]) -> None:
def validate_config(config: AutorouteConfig, discovered: tuple[DiscoveredModel, ...]) -> None:
"""Raise ConfigGenerationError if config references a model discovery didn't return."""
chat_names: FrozenSet[str] = frozenset(m.name for m in chat_models(discovered))
embedding_names: FrozenSet[str] = frozenset(m.name for m in embedding_models(discovered))
chat_names: frozenset[str] = frozenset(m.name for m in chat_models(discovered))
embedding_names: frozenset[str] = frozenset(m.name for m in embedding_models(discovered))
for tier, models in config.tiers.items():
for model in models:
@ -138,7 +138,7 @@ def validate_config(config: AutorouteConfig, discovered: Tuple[DiscoveredModel,
)
def _litellm_proxy_deployment(name: str, base_url: str, api_key: str) -> Dict[str, JsonValue]:
def _litellm_proxy_deployment(name: str, base_url: str, api_key: str) -> dict[str, JsonValue]:
return {
"model_name": name,
"litellm_params": {
@ -149,7 +149,7 @@ def _litellm_proxy_deployment(name: str, base_url: str, api_key: str) -> Dict[st
}
def build_generated_model_list(config: AutorouteConfig) -> List[JsonValue]:
def build_generated_model_list(config: AutorouteConfig) -> list[JsonValue]:
"""Build the model_list for the ephemeral proxy's config.yaml.
Every real model referenced anywhere (tier targets, classifier, embedding) is deduplicated
@ -167,7 +167,7 @@ def build_generated_model_list(config: AutorouteConfig) -> List[JsonValue]:
_litellm_proxy_deployment(name, config.base_url, config.api_key) for name in sorted(referenced_names)
]
complexity_router_config: Dict[str, JsonValue] = {
complexity_router_config: dict[str, JsonValue] = {
"tiers": {tier: list(models) for tier, models in config.tiers.items()},
"default_model": config.default_model,
}
@ -185,7 +185,7 @@ def build_generated_model_list(config: AutorouteConfig) -> List[JsonValue]:
if config.adaptive:
complexity_router_config["adaptive"] = True
auto_router_litellm_params: Dict[str, JsonValue] = {
auto_router_litellm_params: dict[str, JsonValue] = {
"model": "auto_router/complexity_router",
"complexity_router_config": complexity_router_config,
}
@ -202,7 +202,7 @@ def build_generated_model_list(config: AutorouteConfig) -> List[JsonValue]:
]
def build_generated_proxy_config(config: AutorouteConfig, master_key: str) -> Dict[str, JsonValue]:
def build_generated_proxy_config(config: AutorouteConfig, master_key: str) -> dict[str, JsonValue]:
"""Full config.yaml content for the ephemeral proxy, including its own auth key.
master_key must live under general_settings, not litellm_settings -- the proxy server
@ -217,20 +217,20 @@ def build_generated_proxy_config(config: AutorouteConfig, master_key: str) -> Di
__all__ = [
"TIER_NAMES",
"AUTOROUTER_MODEL_NAME",
"TIER_NAMES",
"AutorouteConfig",
"ClassifierChoice",
"ConfigGenerationError",
"DiscoveredModel",
"parse_discovered_models",
"chat_models",
"embedding_models",
"HeuristicClassifier",
"LLMClassifier",
"ClassifierChoice",
"NoSemanticMatching",
"SemanticMatching",
"SemanticMatchingChoice",
"AutorouteConfig",
"validate_config",
"build_generated_model_list",
"chat_models",
"embedding_models",
"parse_discovered_models",
"validate_config",
]

19
litellm/proxy/client/cli/commands/autoroute/process.py
View file

@ -9,7 +9,6 @@ import threading
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Optional
import click
import requests
@ -76,7 +75,7 @@ def poll_liveliness(base_url: str, log_path: Path, process: "subprocess.Popen[by
)
def write_pid_record(record: PidRecord, path: Optional[Path] = None) -> None:
def write_pid_record(record: PidRecord, path: Path | None = None) -> None:
resolved_path = path if path is not None else PID_RECORD_PATH
resolved_path.parent.mkdir(parents=True, exist_ok=True)
with open(resolved_path, "w") as f:
@ -87,7 +86,7 @@ def write_pid_record(record: PidRecord, path: Optional[Path] = None) -> None:
)
def read_pid_record(path: Optional[Path] = None) -> Optional[PidRecord]:
def read_pid_record(path: Path | None = None) -> PidRecord | None:
resolved_path = path if path is not None else PID_RECORD_PATH
if not resolved_path.exists():
return None
@ -95,7 +94,7 @@ def read_pid_record(path: Optional[Path] = None) -> Optional[PidRecord]:
return _PID_RECORD_ADAPTER.validate_json(f.read())
def clear_pid_record(path: Optional[Path] = None) -> None:
def clear_pid_record(path: Path | None = None) -> None:
resolved_path = path if path is not None else PID_RECORD_PATH
resolved_path.unlink(missing_ok=True)
@ -144,15 +143,15 @@ __all__ = [
"CONFIG_PATH",
"LOG_PATH",
"PID_RECORD_PATH",
"ProcessLaunchError",
"PidRecord",
"ProcessLaunchError",
"allocate_free_port",
"launch_proxy",
"poll_liveliness",
"write_pid_record",
"read_pid_record",
"clear_pid_record",
"is_running",
"terminate",
"launch_proxy",
"poll_liveliness",
"read_pid_record",
"stream_log",
"terminate",
"write_pid_record",
]

10
litellm/proxy/client/cli/commands/autoroute/settings.py
View file

@ -1,5 +1,3 @@
from typing import Dict
from pydantic import JsonValue
from .config import AUTOROUTER_MODEL_NAME
@ -22,8 +20,8 @@ ANTHROPIC_DEFAULT_MODEL_ENV_KEYS = (
def merge_claude_settings_static_token(
settings: Dict[str, JsonValue], base_url: str, auth_token: str
) -> Dict[str, JsonValue]:
settings: dict[str, JsonValue], base_url: str, auth_token: str
) -> dict[str, JsonValue]:
"""Return a new settings dict wired to a local ephemeral proxy with a static token.
Unlike up.py's merge_claude_settings (which sets apiKeyHelper for a long-lived, real
@ -33,14 +31,14 @@ def merge_claude_settings_static_token(
"""
raw_env = settings.get(ENV_KEY, {})
base_env = raw_env if isinstance(raw_env, dict) else {}
env: Dict[str, JsonValue] = {
env: dict[str, JsonValue] = {
**base_env,
ANTHROPIC_BASE_URL_KEY: base_url.rstrip("/"),
ANTHROPIC_AUTH_TOKEN_KEY: auth_token,
**{key: AUTOROUTER_MODEL_NAME for key in ANTHROPIC_DEFAULT_MODEL_ENV_KEYS},
}
env.pop(ANTHROPIC_API_KEY_KEY, None)
merged: Dict[str, JsonValue] = {**settings, ENV_KEY: env}
merged: dict[str, JsonValue] = {**settings, ENV_KEY: env}
merged.pop(API_KEY_HELPER_KEY, None)
return merged

7
litellm/proxy/client/cli/commands/autoroute/wizard.py
View file

@ -1,6 +1,5 @@
import sys
from pathlib import Path
from typing import List, Tuple
import click
import yaml
@ -30,7 +29,7 @@ def _is_interactive() -> bool:
return sys.stdin.isatty()
def _fuzzy_pick(models: Tuple[DiscoveredModel, ...], prompt_label: str, multiselect: bool) -> List[str]:
def _fuzzy_pick(models: tuple[DiscoveredModel, ...], prompt_label: str, multiselect: bool) -> list[str]:
"""Type-to-filter picker over a (possibly huge) model pool, using InquirerPy's fzf-style fuzzy prompt.
A plain numbered table + typed index does not scale past a handful of models -- proxies with
@ -56,11 +55,11 @@ def _fuzzy_pick(models: Tuple[DiscoveredModel, ...], prompt_label: str, multisel
click.echo("Select at least one model.")
def _render_and_prompt_for_model(models: Tuple[DiscoveredModel, ...], prompt_label: str) -> str:
def _render_and_prompt_for_model(models: tuple[DiscoveredModel, ...], prompt_label: str) -> str:
return _fuzzy_pick(models, prompt_label, multiselect=False)[0]
def _render_and_prompt_for_models(models: Tuple[DiscoveredModel, ...], prompt_label: str) -> Tuple[str, ...]:
def _render_and_prompt_for_models(models: tuple[DiscoveredModel, ...], prompt_label: str) -> tuple[str, ...]:
return tuple(_fuzzy_pick(models, prompt_label, multiselect=True))

34
litellm/proxy/client/cli/commands/up.py
View file

@ -9,7 +9,7 @@ import threading
from dataclasses import dataclass
from pathlib import Path
from types import FrameType
from typing import Dict, Mapping, Optional
from typing import Mapping
import click
from pydantic import JsonValue, TypeAdapter
@ -37,14 +37,14 @@ class BackupRecord:
"""Snapshot of ~/.claude/settings.json taken right before `lite up` patches it."""
existed: bool
content: Optional[Dict[str, JsonValue]]
content: dict[str, JsonValue] | None
_SETTINGS_ADAPTER = TypeAdapter(Dict[str, JsonValue])
_SETTINGS_ADAPTER = TypeAdapter(dict[str, JsonValue])
_BACKUP_RECORD_ADAPTER = TypeAdapter(BackupRecord)
def load_json_or_empty(path: Path) -> Dict[str, JsonValue]:
def load_json_or_empty(path: Path) -> dict[str, JsonValue]:
if not path.exists():
return {}
with open(path, "r") as f:
@ -56,7 +56,7 @@ def load_json_or_empty(path: Path) -> Dict[str, JsonValue]:
def merge_claude_settings(
settings: Mapping[str, JsonValue], base_url: str, api_key_helper: str
) -> Dict[str, JsonValue]:
) -> dict[str, JsonValue]:
"""Return a new settings dict wired to route Claude Code through the proxy.
Only env.ANTHROPIC_BASE_URL and the top-level apiKeyHelper are overridden; a
@ -71,7 +71,7 @@ def merge_claude_settings(
return {**settings, ENV_KEY: env, API_KEY_HELPER_KEY: api_key_helper}
def write_backup(record: BackupRecord, backup_path: Optional[Path] = None) -> None:
def write_backup(record: BackupRecord, backup_path: Path | None = None) -> None:
path = backup_path if backup_path is not None else BACKUP_PATH
path.parent.mkdir(exist_ok=True)
with open(path, "w") as f:
@ -79,7 +79,7 @@ def write_backup(record: BackupRecord, backup_path: Optional[Path] = None) -> No
os.chmod(path, 0o600)
def read_backup(backup_path: Optional[Path] = None) -> Optional[BackupRecord]:
def read_backup(backup_path: Path | None = None) -> BackupRecord | None:
path = backup_path if backup_path is not None else BACKUP_PATH
if not path.exists():
return None
@ -87,9 +87,7 @@ def read_backup(backup_path: Optional[Path] = None) -> Optional[BackupRecord]:
return _BACKUP_RECORD_ADAPTER.validate_json(f.read())
def restore_claude_settings(
settings_path: Optional[Path] = None, backup_path: Optional[Path] = None
) -> Optional[BackupRecord]:
def restore_claude_settings(settings_path: Path | None = None, backup_path: Path | None = None) -> BackupRecord | None:
"""Restore settings_path from the backup at backup_path, then delete the backup.
Returns the restored record, or None if there was nothing to restore.
@ -199,7 +197,7 @@ def up(ctx: click.Context) -> None:
stop_event = threading.Event()
restored = threading.Lock()
def _handle_signal(_signum: int, _frame: Optional[FrameType]) -> None:
def _handle_signal(_signum: int, _frame: FrameType | None) -> None:
stop_event.set()
def _restore_once() -> None:
@ -225,16 +223,16 @@ def down() -> None:
__all__ = [
"up",
"down",
"BACKUP_PATH",
"CLAUDE_SETTINGS_PATH",
"BackupRecord",
"UpError",
"down",
"load_json_or_empty",
"merge_claude_settings",
"write_backup",
"read_backup",
"restore_claude_settings",
"resolve_api_key_helper",
"UpError",
"CLAUDE_SETTINGS_PATH",
"BACKUP_PATH",
"restore_claude_settings",
"up",
"write_backup",
]