scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-06 02:48:13 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 128

fix(lint): replace Any with proper types, use datetime.now(timezone.utc)

Browse source 
Fix strict-budget violations introduced by new files:
- Replace typing.Any with object/proper types in callback signatures,
  guardrail hooks, usage endpoints, and common request processing
- Use datetime.now(timezone.utc) instead of datetime.now() for
  timezone-aware datetimes in bedrock_guardrails and bias_hallucination_estimator
- Remove noqa suppressions in favour of actual fixes

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Sameer Kankute 2026-06-24 15:57:10 +05:30
parent e2c31fd471
commit 41abaa7aad
No known key found for this signature in database
5 changed files with 66 additions and 60 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

38
litellm/integrations/asqav/asqav.py
View file

@ -63,7 +63,7 @@ def _read_tail(fh: BinaryIO, size: int) -> bytes:
chunk_size *= 2
def _content_digest(value: Any) -> Optional[str]: # noqa: ANN401
def _content_digest(value: object) -> Optional[str]:
"""Return a SHA-256 hex digest of a content value, or None if empty."""
if value is None:
return None
@ -73,9 +73,9 @@ def _content_digest(value: Any) -> Optional[str]: # noqa: ANN401
def _extract_loggable(
kwargs: dict[str, Any],
response_obj: Any, # noqa: ANN401
start_time: Any, # noqa: ANN401
end_time: Any, # noqa: ANN401
response_obj: object,
start_time: Optional[datetime],
end_time: Optional[datetime],
status: str,
) -> dict[str, Any]:
"""Pull metadata + digests out of a callback invocation.
@ -276,9 +276,9 @@ class AsqavLogger(CustomLogger):
def _build_and_append(
self,
kwargs: dict[str, Any],
response_obj: Any, # noqa: ANN401
start_time: Any, # noqa: ANN401
end_time: Any, # noqa: ANN401
response_obj: object,
start_time: Optional[datetime],
end_time: Optional[datetime],
status: str,
) -> None:
"""Build one audit record and append it to the JSONL log.
@ -372,27 +372,27 @@ class AsqavLogger(CustomLogger):
def log_success_event(
self,
kwargs: dict[str, Any],
response_obj: Any,
start_time: Any,
end_time: Any, # noqa: ANN401
response_obj: object,
start_time: Optional[datetime],
end_time: Optional[datetime],
) -> None:
self._build_and_append(kwargs, response_obj, start_time, end_time, "success")
def log_failure_event(
self,
kwargs: dict[str, Any],
response_obj: Any,
start_time: Any,
end_time: Any, # noqa: ANN401
response_obj: object,
start_time: Optional[datetime],
end_time: Optional[datetime],
) -> None:
self._build_and_append(kwargs, response_obj, start_time, end_time, "failure")
async def async_log_success_event(
self,
kwargs: dict[str, Any],
response_obj: Any,
start_time: Any,
end_time: Any, # noqa: ANN401
response_obj: object,
start_time: Optional[datetime],
end_time: Optional[datetime],
) -> None:
await asyncio.to_thread(
self._build_and_append,
@ -406,9 +406,9 @@ class AsqavLogger(CustomLogger):
async def async_log_failure_event(
self,
kwargs: dict[str, Any],
response_obj: Any,
start_time: Any,
end_time: Any, # noqa: ANN401
response_obj: object,
start_time: Optional[datetime],
end_time: Optional[datetime],
) -> None:
await asyncio.to_thread(
self._build_and_append,

2
litellm/proxy/common_request_processing.py
View file

@ -2437,7 +2437,7 @@ class ProxyBaseLLMRequestProcessing:
@staticmethod
async def _bill_partial_stream_on_disconnect(
response: Any, request_data: dict # noqa: ANN401
response: object, request_data: dict
) -> None:
"""Record SpendLogs for tokens already produced when a stream is cut off.

46
litellm/proxy/guardrails/guardrail_hooks/bedrock_guardrails.py
View file

@ -184,7 +184,7 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
guardrailIdentifier: Optional[str] = None,
guardrailVersion: Optional[str] = None,
disable_exception_on_block: Optional[bool] = False,
checks: Any | None = None, # noqa: ANN401
checks: dict | object | None = None,
content_filter_threshold: float | None = 0.5,
prompt_attack_threshold: float | None = 0.5,
pii_confidence_threshold: float | None = 0.5,
@ -253,7 +253,7 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
)
@staticmethod
def _normalize_checks(checks: Any | None) -> dict[str, Any] | None: # noqa: ANN401
def _normalize_checks(checks: object | None) -> dict[str, Any] | None:
"""Normalize the configured `checks` into a plain dict for the API body.
Accepts a pydantic ``BedrockChecksConfigModel`` or a raw dict; drops empty /
@ -729,7 +729,7 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
self,
source: Literal["INPUT", "OUTPUT"],
messages: list[AllMessageValues] | None = None,
response: Any | litellm.ModelResponse | None = None, # noqa: ANN401
response: litellm.ModelResponse | None = None,
request_data: dict | None = None,
logging_event_type: GuardrailEventHooks | None = None,
) -> BedrockGuardrailResponse:
@ -760,13 +760,13 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
self,
source: Literal["INPUT", "OUTPUT"],
messages: list[AllMessageValues] | None = None,
response: Any | litellm.ModelResponse | None = None, # noqa: ANN401
response: litellm.ModelResponse | None = None,
request_data: dict | None = None,
logging_event_type: GuardrailEventHooks | None = None,
) -> BedrockGuardrailResponse:
from datetime import datetime
from datetime import datetime, timezone
start_time = datetime.now()
start_time = datetime.now(timezone.utc)
credentials, aws_region_name = self._load_credentials()
bedrock_request_data: dict = dict(
self.convert_to_bedrock_format(source=source, messages=messages, response=response)
@ -828,8 +828,8 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
request_data=request_data or {},
guardrail_status=self._get_bedrock_guardrail_response_status(response=httpx_response),
start_time=start_time.timestamp(),
end_time=datetime.now().timestamp(), # noqa: DTZ005
duration=(datetime.now() - start_time).total_seconds(), # noqa: DTZ005
end_time=datetime.now(timezone.utc).timestamp(),
duration=(datetime.now(timezone.utc) - start_time).total_seconds(),
event_type=event_type,
tracing_detail=tracing_detail or None,
)
@ -856,7 +856,7 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
async def _sign_and_post(
self,
prepared_request: Any, # noqa: ANN401
prepared_request: object,
request_data: dict | None,
event_type: GuardrailEventHooks,
start_time: "datetime",
@ -868,7 +868,7 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
success (including non-2xx that httpx did not raise on); the 200-path logging,
status and tracing stay with each caller because the two APIs report differently.
"""
from datetime import datetime
from datetime import datetime, timezone
try:
return await self.async_handler.post(
@ -895,8 +895,10 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
request_data=request_data or {},
guardrail_status="guardrail_failed_to_respond",
start_time=start_time.timestamp(),
end_time=datetime.now().timestamp(),
duration=(datetime.now() - start_time).total_seconds(),
end_time=datetime.now(timezone.utc).timestamp(),
duration=(
datetime.now(timezone.utc) - start_time
).total_seconds(),
event_type=event_type,
)
raise HTTPException(
@ -914,8 +916,8 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
request_data=request_data or {},
guardrail_status="guardrail_failed_to_respond",
start_time=start_time.timestamp(),
end_time=datetime.now().timestamp(),
duration=(datetime.now() - start_time).total_seconds(),
end_time=datetime.now(timezone.utc).timestamp(),
duration=(datetime.now(timezone.utc) - start_time).total_seconds(),
event_type=event_type,
)
raise
@ -945,7 +947,7 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
self,
source: Literal["INPUT", "OUTPUT"],
messages: list[AllMessageValues] | None = None,
response: Any | litellm.ModelResponse | None = None, # noqa: ANN401
response: litellm.ModelResponse | None = None,
) -> list[BedrockChecksMessage]:
"""Build the role-tagged `messages` array for InvokeGuardrailChecks.
@ -985,7 +987,7 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
self,
source: Literal["INPUT", "OUTPUT"],
messages: list[AllMessageValues] | None = None,
response: Any | litellm.ModelResponse | None = None, # noqa: ANN401
response: litellm.ModelResponse | None = None,
request_data: dict | None = None,
logging_event_type: GuardrailEventHooks | None = None,
) -> BedrockGuardrailResponse:
@ -995,9 +997,9 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
to a block decision via the configured thresholds. On a pass we return an
empty ``BedrockGuardrailResponse`` (downstream masking treats it as a no-op).
"""
from datetime import datetime
from datetime import datetime, timezone
start_time = datetime.now() # noqa: DTZ005
start_time = datetime.now(timezone.utc)
checks_messages = self._build_invoke_guardrail_checks_messages(
source=source, messages=messages, response=response
@ -1050,8 +1052,8 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
request_data=request_data or {},
guardrail_status="guardrail_failed_to_respond",
start_time=start_time.timestamp(),
end_time=datetime.now().timestamp(), # noqa: DTZ005
duration=(datetime.now() - start_time).total_seconds(), # noqa: DTZ005
end_time=datetime.now(timezone.utc).timestamp(),
duration=(datetime.now(timezone.utc) - start_time).total_seconds(),
event_type=event_type,
)
raise HTTPException(status_code=status_code, detail=detail_message)
@ -1071,8 +1073,8 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
request_data=request_data or {},
guardrail_status=self._get_invoke_checks_status(bool(violations)),
start_time=start_time.timestamp(),
end_time=datetime.now().timestamp(), # noqa: DTZ005
duration=(datetime.now() - start_time).total_seconds(), # noqa: DTZ005
end_time=datetime.now(timezone.utc).timestamp(),
duration=(datetime.now(timezone.utc) - start_time).total_seconds(),
event_type=event_type,
tracing_detail=self._build_invoke_checks_tracing_detail(violations) or None,
)

8
litellm/proxy/guardrails/guardrail_hooks/bias_hallucination_estimator/bias_hallucination_estimator.py
View file

@ -1,7 +1,7 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
from datetime import datetime, timezone
from typing import (
TYPE_CHECKING,
Literal,
@ -150,7 +150,7 @@ class BiasHallucinationEstimatorGuardrail(CustomGuardrail):
if not self._should_check(input_type):
return inputs
start_time = datetime.now() # noqa: DTZ005
start_time = datetime.now(timezone.utc)
texts = self._extract_texts(inputs)
if not texts:
return inputs
@ -274,8 +274,8 @@ class BiasHallucinationEstimatorGuardrail(CustomGuardrail):
request_data=request_data,
guardrail_status=status,
start_time=start_time.timestamp(),
end_time=datetime.now().timestamp(), # noqa: DTZ005
duration=(datetime.now() - start_time).total_seconds(),
end_time=datetime.now(timezone.utc).timestamp(),
duration=(datetime.now(timezone.utc) - start_time).total_seconds(),
tracing_detail=tracing_detail,
)

32
litellm/proxy/guardrails/usage_endpoints.py
View file

@ -101,7 +101,9 @@ def _trend_from_comparison(current_fail: float, previous_fail: float) -> str:
return "stable"
def _aggregate_daily_metrics(metrics: Any, id_attr: str) -> Dict[str, Dict[str, Any]]:
def _aggregate_daily_metrics(
metrics: object, id_attr: str
) -> Dict[str, Dict[str, Any]]:
agg: Dict[str, Dict[str, Any]] = {}
for m in metrics:
gid = getattr(m, id_attr)
@ -114,7 +116,7 @@ def _aggregate_daily_metrics(metrics: Any, id_attr: str) -> Dict[str, Dict[str,
return agg
def _prev_fail_rates(metrics_prev: Any, id_attr: str) -> Dict[str, float]:
def _prev_fail_rates(metrics_prev: object, id_attr: str) -> Dict[str, float]:
prev_agg_raw: Dict[str, Dict[str, int]] = {}
for m in metrics_prev:
gid = getattr(m, id_attr)
@ -126,7 +128,7 @@ def _prev_fail_rates(metrics_prev: Any, id_attr: str) -> Dict[str, float]:
return {gid: (100.0 * v["blocked"] / v["req"]) if v["req"] else 0.0 for gid, v in prev_agg_raw.items()}
def _chart_from_metrics(metrics: Any) -> List[Dict[str, Any]]:
def _chart_from_metrics(metrics: object) -> List[Dict[str, Any]]:
chart_by_date: Dict[str, Dict[str, int]] = {}
for m in metrics:
d = m.date
@ -137,20 +139,20 @@ def _chart_from_metrics(metrics: Any) -> List[Dict[str, Any]]:
return [{"date": d, "passed": v["passed"], "blocked": v["blocked"]} for d, v in sorted(chart_by_date.items())]
def _get_guardrail_attrs(g: Any) -> tuple[Any, str]:
def _get_guardrail_attrs(g: object) -> tuple[object, str]:
"""Get (guardrail_id, display_name) from guardrail - handles Prisma model or dict."""
gid = getattr(g, "guardrail_id", None) or (g.get("guardrail_id") if isinstance(g, dict) else None)
name = getattr(g, "guardrail_name", None) or (g.get("guardrail_name") if isinstance(g, dict) else None)
return gid, (name or gid or "")
def _get_guardrail_dict_field(g: Any, field_name: str) -> Any: # noqa: ANN401
def _get_guardrail_dict_field(g: object, field_name: str) -> object:
if isinstance(g, dict):
return g.get(field_name)
return getattr(g, field_name, None)
def _get_guardrail_litellm_params(g: Any) -> dict[str, Any]: # noqa: ANN401
def _get_guardrail_litellm_params(g: object) -> dict[str, Any]:
litellm_params = _get_guardrail_dict_field(g, "litellm_params")
if isinstance(litellm_params, dict):
return litellm_params
@ -159,7 +161,7 @@ def _get_guardrail_litellm_params(g: Any) -> dict[str, Any]: # noqa: ANN401
return {}
def _get_guardrail_info(g: Any) -> dict[str, Any]: # noqa: ANN401
def _get_guardrail_info(g: object) -> dict[str, Any]:
guardrail_info = _get_guardrail_dict_field(g, "guardrail_info")
return guardrail_info if isinstance(guardrail_info, dict) else {}
@ -177,7 +179,7 @@ def _get_config_loaded_guardrails() -> list[Any]:
def _find_config_loaded_guardrail(
guardrail_id_or_name: str,
) -> Optional[Any]: # noqa: ANN401
) -> Optional[object]:
for guardrail in _get_config_loaded_guardrails():
gid, display_name = _get_guardrail_attrs(guardrail)
if guardrail_id_or_name in (gid, display_name):
@ -185,7 +187,7 @@ def _find_config_loaded_guardrail(
return None
def _merge_config_loaded_guardrails(db_guardrails: Any) -> list[Any]: # noqa: ANN401
def _merge_config_loaded_guardrails(db_guardrails: object) -> list[object]:
guardrails = list(db_guardrails)
seen_keys: set[str] = set()
for guardrail in guardrails:
@ -203,7 +205,7 @@ def _merge_config_loaded_guardrails(db_guardrails: Any) -> list[Any]: # noqa: A
def _guardrail_overview_rows(
guardrails: Any,
guardrails: object,
agg: Dict[str, Dict[str, Any]],
prev_agg: Dict[str, float],
) -> List[UsageOverviewRow]:
@ -271,7 +273,7 @@ def _guardrail_overview_rows(
def _policy_overview_rows(
policies: Any,
policies: object,
agg: Dict[str, Dict[str, Any]],
prev_agg: Dict[str, float],
) -> List[UsageOverviewRow]:
@ -476,7 +478,9 @@ def _build_usage_logs_where(
return where
def _usage_log_entry_from_row(r: Any, sl: Any, action_filter: Optional[str]) -> Optional[UsageLogEntry]:
def _usage_log_entry_from_row(
r: object, sl: object, action_filter: Optional[str]
) -> Optional[UsageLogEntry]:
meta = sl.metadata
if isinstance(meta, str):
try:
@ -526,7 +530,7 @@ def _usage_log_entry_from_row(r: Any, sl: Any, action_filter: Optional[str]) ->
)
def _snippet(text: Any, max_len: int = 200) -> Optional[str]:
def _snippet(text: object, max_len: int = 200) -> Optional[str]:
if text is None:
return None
if isinstance(text, str):
@ -548,7 +552,7 @@ def _snippet(text: Any, max_len: int = 200) -> Optional[str]:
return result
def _input_snippet_for_log(sl: Any) -> Optional[str]:
def _input_snippet_for_log(sl: object) -> Optional[str]:
"""Snippet for request input: prefer messages, fall back to proxy_server_request (same as drawer)."""
out = _snippet(sl.messages)
if out: