style: fix linting and formatting issues

This commit is contained in:
Devdeep Sarkar 2026-06-03 20:08:26 +05:30
parent 896a093399
commit a0c54c7f22
5 changed files with 17 additions and 10 deletions

View file

@ -18,7 +18,6 @@ from datetime import datetime, timedelta
from headroom import paths as _paths
from headroom.pricing.litellm_pricing import resolve_litellm_model
log = logging.getLogger(__name__)
LOG_DIR = _paths.log_dir()

View file

@ -44,6 +44,8 @@ _MODEL_ALIASES: dict[str, str] = {
}
_resolved_model_cache: dict[str, str] = {}
def resolve_litellm_model(model: str) -> str:
"""Resolve model name to one LiteLLM recognizes, adding provider prefix if needed.
Results are cached per model name to avoid blocking the event loop
@ -54,6 +56,8 @@ def resolve_litellm_model(model: str) -> str:
resolved = _resolve_litellm_model_uncached(model)
_resolved_model_cache[model] = resolved
return resolved
def _resolve_litellm_model_uncached(model: str) -> str:
"""Uncached resolution — called once per unique model name."""
if not LITELLM_AVAILABLE:
@ -82,7 +86,7 @@ def _resolve_litellm_model_uncached(model: str) -> str:
except Exception:
break
return model
@dataclass
class LiteLLMModelPricing:

View file

@ -16,7 +16,6 @@ from typing import TYPE_CHECKING, Any
from headroom.proxy.modes import PROXY_MODE_CACHE
if TYPE_CHECKING:
from headroom.proxy.prometheus_metrics import PrometheusMetrics
@ -567,7 +566,6 @@ class CostTracker:
self._api_cache_write_1h_by_model.clear()
self._api_uncached_by_model.clear()
def estimate_cost(
self,
model: str,
@ -595,6 +593,7 @@ class CostTracker:
try:
from headroom.pricing.litellm_pricing import resolve_litellm_model
resolved_model = resolve_litellm_model(model)
# litellm.cost_per_token handles all token types natively:
@ -704,6 +703,7 @@ class CostTracker:
return None
try:
from headroom.pricing.litellm_pricing import resolve_litellm_model
resolved = resolve_litellm_model(model)
info = litellm.model_cost.get(resolved, {})
cost_per_token = info.get("input_cost_per_token")
@ -722,6 +722,7 @@ class CostTracker:
return None
try:
from headroom.pricing.litellm_pricing import resolve_litellm_model
resolved = resolve_litellm_model(model)
info = litellm.model_cost.get(resolved, {})
uncached = info.get("input_cost_per_token")

View file

@ -37,6 +37,7 @@ def test_savings_at_list_price():
# Savings should be 100k tokens * list input price (NOT affected by cache mix)
import litellm
from headroom.pricing.litellm_pricing import resolve_litellm_model
resolved = resolve_litellm_model(model)

View file

@ -33,7 +33,8 @@ class TestModelResolutionCaching:
import headroom.pricing.litellm_pricing as lp
with patch(
"headroom.pricing.litellm_pricing._resolve_litellm_model_uncached", return_value="anthropic/claude-opus-4-6"
"headroom.pricing.litellm_pricing._resolve_litellm_model_uncached",
return_value="anthropic/claude-opus-4-6",
) as mock_uncached:
# First call — should invoke uncached resolution
result1 = lp.resolve_litellm_model("claude-opus-4-6")
@ -165,7 +166,8 @@ class TestModelResolutionCaching:
import headroom.pricing.litellm_pricing as lp
with patch(
"headroom.pricing.litellm_pricing._resolve_litellm_model_uncached", return_value="resolved/model-a"
"headroom.pricing.litellm_pricing._resolve_litellm_model_uncached",
return_value="resolved/model-a",
) as mock_uncached:
# Resolve
result1 = lp.resolve_litellm_model("model-a")
@ -417,12 +419,12 @@ class TestConcurrentSessionSafety:
return f"resolved/{model}"
with patch(
"headroom.pricing.litellm_pricing._resolve_litellm_model_uncached", side_effect=slow_uncached
"headroom.pricing.litellm_pricing._resolve_litellm_model_uncached",
side_effect=slow_uncached,
):
# Launch 50 concurrent resolution tasks for the same model
tasks = [
asyncio.to_thread(lp.resolve_litellm_model, "claude-opus-4-6")
for _ in range(50)
asyncio.to_thread(lp.resolve_litellm_model, "claude-opus-4-6") for _ in range(50)
]
results = await asyncio.gather(*tasks)
@ -503,8 +505,8 @@ class TestConcurrentSessionSafety:
@pytest.mark.asyncio
async def test_estimate_cost_concurrent_with_caching(self):
"""Multiple concurrent estimate_cost calls should not block each other."""
from headroom.proxy.server import CostTracker
import headroom.pricing.litellm_pricing as lp
from headroom.proxy.server import CostTracker
tracker = CostTracker()