Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/guardrails/usage_endpoints.py: 55%

1""" 

2Guardrails and policies usage endpoints for the dashboard. 

3GET /guardrails/usage/overview, /guardrails/usage/detail/:id, /guardrails/usage/logs 

4""" 

5 

6import json 

7from collections.abc import Callable, Iterable, Mapping, Sequence 

8from datetime import date, datetime, timedelta, timezone 

9from itertools import groupby 

10from types import MappingProxyType 

11from typing import TYPE_CHECKING, Any, Final, Literal, TypeVar, overload 

12 

13from fastapi import APIRouter, Depends, Query 

14from pydantic import BaseModel, Field 

15from typing_extensions import NotRequired, ReadOnly, TypedDict 

16 

17from litellm._logging import verbose_proxy_logger 

18from litellm.proxy._types import UserAPIKeyAuth 

19from litellm.proxy.auth.user_api_key_auth import user_api_key_auth 

20from litellm.proxy.guardrails.usage_tracking import guardrail_status_to_action 

21from litellm.repositories.prisma_protocols import TableActions 

22from litellm.repositories.table_repositories import ( 

23 DailyGuardrailMetricsRepository, 

24 DailyGuardrailUsageUnitsRepository, 

25 DailyPolicyMetricsRepository, 

26 GuardrailsRepository, 

27 PolicyRepository, 

28 SpendLogGuardrailIndexRepository, 

29 SpendLogsRepository, 

30) 

31 

32if TYPE_CHECKING: 32 ↛ 33line 32 didn't jump to line 33 because the condition on line 32 was never true

33 from prisma import models as prisma_models 

34 from prisma import types as prisma_types 

35 

36 from litellm.proxy.utils import PrismaClient 

37 from litellm.types.guardrails import Guardrail 

38 

39 _DbOrConfigGuardrail = prisma_models.LiteLLM_GuardrailsTable | Guardrail 

40 _DailyMetricsRow = prisma_models.LiteLLM_DailyGuardrailMetrics | prisma_models.LiteLLM_DailyPolicyMetrics 

41 

42router: Final = APIRouter() 

43 

44_EMPTY_UNITS: Final[Mapping[str, int]] = MappingProxyType({}) 

45_ACTION_SEVERITY: Final[Mapping[str, int]] = MappingProxyType({"not_run": 0, "passed": 1, "flagged": 2, "blocked": 3}) 

46 

47_T = TypeVar("_T") 

48_MetricsRowT = TypeVar("_MetricsRowT", bound="_DailyMetricsRow") 

49 

50_USAGE_MAX_RANGE_DAYS: Final = 366 

51 

52 

53def _resolve_usage_window(start_date: str | None, end_date: str | None) -> tuple[str, str]: 

54 from fastapi import HTTPException, status 

55 

56 now: Final = datetime.now(timezone.utc) 

57 end: Final = end_date or now.strftime("%Y-%m-%d") 

58 start: Final = start_date or (now - timedelta(days=7)).strftime("%Y-%m-%d") 

59 try: 

60 parsed: Final = (date.fromisoformat(start), date.fromisoformat(end)) 

61 except ValueError: 

62 raise HTTPException( 

63 status_code=status.HTTP_400_BAD_REQUEST, 

64 detail="start_date and end_date must be in YYYY-MM-DD format", 

65 ) 

66 start_obj, end_obj = parsed 

67 if (start_obj.isoformat(), end_obj.isoformat()) != (start, end): 67 ↛ 68line 67 didn't jump to line 68 because the condition on line 67 was never true

68 raise HTTPException( 

69 status_code=status.HTTP_400_BAD_REQUEST, 

70 detail="start_date and end_date must be in YYYY-MM-DD format", 

71 ) 

72 if end_obj < start_obj: 72 ↛ 73line 72 didn't jump to line 73 because the condition on line 72 was never true

73 raise HTTPException( 

74 status_code=status.HTTP_400_BAD_REQUEST, 

75 detail="start_date must be on or before end_date", 

76 ) 

77 if end_obj - start_obj > timedelta(days=_USAGE_MAX_RANGE_DAYS): 77 ↛ 78line 77 didn't jump to line 78 because the condition on line 77 was never true

78 raise HTTPException( 

79 status_code=status.HTTP_400_BAD_REQUEST, 

80 detail=f"Date range too large; maximum is {_USAGE_MAX_RANGE_DAYS} days", 

81 ) 

82 return start, end 

83 

84 

85def _guardrails_table( 

86 prisma_client: "PrismaClient", 

87) -> "TableActions[prisma_models.LiteLLM_GuardrailsTable]": 

88 guardrails_table: Final[TableActions[prisma_models.LiteLLM_GuardrailsTable]] = GuardrailsRepository( 

89 prisma_client 

90 ).table 

91 return guardrails_table 

92 

93 

94def _policies_table( 

95 prisma_client: "PrismaClient", 

96) -> "TableActions[prisma_models.LiteLLM_PolicyTable]": 

97 policies_table: Final[TableActions[prisma_models.LiteLLM_PolicyTable]] = PolicyRepository(prisma_client).table 

98 return policies_table 

99 

100 

101def _daily_guardrail_metrics_table( 

102 prisma_client: "PrismaClient", 

103) -> "TableActions[prisma_models.LiteLLM_DailyGuardrailMetrics]": 

104 metrics_table: Final[TableActions[prisma_models.LiteLLM_DailyGuardrailMetrics]] = DailyGuardrailMetricsRepository( 

105 prisma_client 

106 ).table 

107 return metrics_table 

108 

109 

110def _daily_policy_metrics_table( 

111 prisma_client: "PrismaClient", 

112) -> "TableActions[prisma_models.LiteLLM_DailyPolicyMetrics]": 

113 metrics_table: Final[TableActions[prisma_models.LiteLLM_DailyPolicyMetrics]] = DailyPolicyMetricsRepository( 

114 prisma_client 

115 ).table 

116 return metrics_table 

117 

118 

119async def _find_daily_guardrail_metrics( 

120 prisma_client: "PrismaClient", 

121 where: "prisma_types.LiteLLM_DailyGuardrailMetricsWhereInput", 

122) -> "Sequence[prisma_models.LiteLLM_DailyGuardrailMetrics]": 

123 return await _daily_guardrail_metrics_table(prisma_client).find_many(where=where) 

124 

125 

126async def _find_daily_policy_metrics( 

127 prisma_client: "PrismaClient", 

128 where: "prisma_types.LiteLLM_DailyPolicyMetricsWhereInput", 

129) -> "Sequence[prisma_models.LiteLLM_DailyPolicyMetrics]": 

130 return await _daily_policy_metrics_table(prisma_client).find_many(where=where) 

131 

132 

133def _daily_guardrail_usage_units_table( 

134 prisma_client: "PrismaClient", 

135) -> "TableActions[prisma_models.LiteLLM_DailyGuardrailUsageUnits]": 

136 units_table: Final[TableActions[prisma_models.LiteLLM_DailyGuardrailUsageUnits]] = ( 

137 DailyGuardrailUsageUnitsRepository(prisma_client).table 

138 ) 

139 return units_table 

140 

141 

142async def _find_daily_guardrail_usage_units( 

143 prisma_client: "PrismaClient", 

144 where: "prisma_types.LiteLLM_DailyGuardrailUsageUnitsWhereInput", 

145) -> "Sequence[prisma_models.LiteLLM_DailyGuardrailUsageUnits]": 

146 from prisma.errors import TableNotFoundError 

147 

148 try: 

149 return await _daily_guardrail_usage_units_table(prisma_client).find_many(where=where) 

150 except TableNotFoundError as e: 

151 verbose_proxy_logger.warning( 

152 "Guardrail usage units are unavailable until the LiteLLM_DailyGuardrailUsageUnits migration is applied: %s", 

153 e, 

154 ) 

155 return () 

156 

157 

158def _counter_name(row: "prisma_models.LiteLLM_DailyGuardrailUsageUnits") -> str: 

159 return row.usage_unit 

160 

161 

162def _team_of(row: "prisma_models.LiteLLM_DailyGuardrailUsageUnits") -> str: 

163 return row.team_id 

164 

165 

166def _key_of(row: "prisma_models.LiteLLM_DailyGuardrailUsageUnits") -> str: 

167 return row.api_key 

168 

169 

170def _row_untracked_units(row: "prisma_models.LiteLLM_DailyGuardrailUsageUnits") -> int: 

171 """A row written before the cost column carries NULL cost and is untracked in full.""" 

172 return int(row.units) if row.cost is None else int(row.untracked_units) 

173 

174 

175def _row_tracked_cost(row: "prisma_models.LiteLLM_DailyGuardrailUsageUnits") -> float | None: 

176 """The row's cost when it prices at least one unit; None when every unit is untracked.""" 

177 return None if row.cost is None or _row_untracked_units(row) >= int(row.units) else row.cost 

178 

179 

180def _sum_counter_units(rows: "Iterable[prisma_models.LiteLLM_DailyGuardrailUsageUnits]") -> Mapping[str, int]: 

181 ordered: Final = sorted(rows, key=_counter_name) 

182 return MappingProxyType( 

183 {name: sum(int(r.units) for r in group) for name, group in groupby(ordered, key=_counter_name)} 

184 ) 

185 

186 

187def _sum_untracked_units(rows: "Iterable[prisma_models.LiteLLM_DailyGuardrailUsageUnits]") -> Mapping[str, int]: 

188 ordered: Final = sorted(rows, key=_counter_name) 

189 per_counter: Final = tuple( 

190 (name, sum(map(_row_untracked_units, group))) for name, group in groupby(ordered, key=_counter_name) 

191 ) 

192 return MappingProxyType({name: units for name, units in per_counter if units}) 

193 

194 

195def _sum_tracked_cost(rows: "Iterable[prisma_models.LiteLLM_DailyGuardrailUsageUnits]") -> float | None: 

196 """Sum over rows that price at least one unit; None when no row does.""" 

197 tracked: Final = tuple(cost for cost in map(_row_tracked_cost, rows) if cost is not None) 

198 return sum(tracked) if tracked else None 

199 

200 

201def _by( 

202 rows: "Sequence[prisma_models.LiteLLM_DailyGuardrailUsageUnits]", 

203 key_of: "Callable[[prisma_models.LiteLLM_DailyGuardrailUsageUnits], str]", 

204 reduce: "Callable[[Iterable[prisma_models.LiteLLM_DailyGuardrailUsageUnits]], _T]", 

205) -> Mapping[str, _T]: 

206 ordered: Final = sorted(rows, key=key_of) 

207 return MappingProxyType({key: reduce(group) for key, group in groupby(ordered, key=key_of)}) 

208 

209 

210def _first_match(lookup_keys: Sequence[str], mapping: Mapping[str, _T], default: _T) -> _T: 

211 return next((mapping[k] for k in lookup_keys if k in mapping), default) 

212 

213 

214# --- Response models --- 

215 

216 

217class _GuardrailRunInfo(TypedDict, total=False): 

218 guardrail_id: ReadOnly[str | None] 

219 guardrail_name: ReadOnly[str | None] 

220 guardrail_status: ReadOnly[str | None] 

221 duration: ReadOnly[float | None] 

222 confidence_score: ReadOnly[float | None] 

223 risk_score: ReadOnly[float | None] 

224 guardrail_response: ReadOnly[str | Mapping[str, object] | Sequence[Mapping[str, object]] | None] 

225 

226 

227class UsageChartPoint(TypedDict): 

228 date: str 

229 passed: int 

230 blocked: int 

231 score: NotRequired[float | None] 

232 

233 

234class _MetricTotals(TypedDict): 

235 requests: int 

236 passed: int 

237 blocked: int 

238 flagged: int 

239 

240 

241class _PrevPeriodCounts(TypedDict): 

242 req: int 

243 blocked: int 

244 

245 

246class _DailyPassBlocked(TypedDict): 

247 passed: int 

248 blocked: int 

249 

250 

251class UsageOverviewRow(BaseModel): 

252 id: str 

253 name: str 

254 type: str 

255 provider: str 

256 requestsEvaluated: int 

257 failRate: float 

258 avgScore: float | None 

259 avgLatency: float | None 

260 status: str # healthy | warning | critical 

261 trend: str # up | down | stable 

262 usageUnits: Mapping[str, int] 

263 cost: float | None = Field( 

264 description="USD for the priced share of usageUnits over the window; null when no unit was priced" 

265 ) 

266 untrackedUsageUnits: Mapping[str, int] = Field( 

267 description="The share of usageUnits that cost leaves out: units recorded with no known price, per counter" 

268 ) 

269 

270 

271class UsageOverviewResponse(BaseModel): 

272 rows: list[UsageOverviewRow] 

273 chart: list[UsageChartPoint] # [{ date, passed, blocked }] 

274 totalRequests: int 

275 totalBlocked: int 

276 passRate: float 

277 totalUsageUnits: Mapping[str, int] 

278 totalCost: float | None 

279 totalUntrackedUsageUnits: Mapping[str, int] 

280 

281 

282_EMPTY_OVERVIEW: Final = UsageOverviewResponse( 

283 rows=[], 

284 chart=[], 

285 totalRequests=0, 

286 totalBlocked=0, 

287 passRate=100.0, 

288 totalUsageUnits=_EMPTY_UNITS, 

289 totalCost=None, 

290 totalUntrackedUsageUnits=_EMPTY_UNITS, 

291) 

292 

293 

294class UsageUnitsDailyPoint(BaseModel): 

295 date: str 

296 units: Mapping[str, int] 

297 cost: float | None 

298 

299 

300class UsageDetailResponse(BaseModel): 

301 guardrail_id: str 

302 guardrail_name: str 

303 type: str 

304 provider: str 

305 requestsEvaluated: int 

306 failRate: float 

307 avgScore: float | None 

308 avgLatency: float | None 

309 status: str 

310 trend: str 

311 description: str | None 

312 time_series: list[UsageChartPoint] 

313 usage_units: Mapping[str, int] 

314 usage_units_daily: Sequence[UsageUnitsDailyPoint] 

315 usage_units_by_team: Mapping[str, Mapping[str, int]] 

316 usage_units_by_key: Mapping[str, Mapping[str, int]] 

317 cost: float | None 

318 cost_by_unit: Mapping[str, float | None] 

319 cost_by_team: Mapping[str, float | None] 

320 cost_by_key: Mapping[str, float | None] 

321 untracked_usage_units: Mapping[str, int] 

322 untracked_usage_units_by_team: Mapping[str, Mapping[str, int]] 

323 untracked_usage_units_by_key: Mapping[str, Mapping[str, int]] 

324 

325 

326class UsageLogEntry(BaseModel): 

327 id: str 

328 timestamp: str 

329 action: str # blocked | passed | flagged | not_run 

330 score: float | None 

331 latency_ms: float | None 

332 model: str | None 

333 input_snippet: str | None 

334 output_snippet: str | None 

335 reason: str | None 

336 

337 

338class UsageLogsResponse(BaseModel): 

339 logs: list[UsageLogEntry] 

340 total: int 

341 page: int 

342 page_size: int 

343 

344 

345def _status_from_fail_rate(fail_rate: float) -> str: 

346 if fail_rate > 15: 346 ↛ 347line 346 didn't jump to line 347 because the condition on line 346 was never true

347 return "critical" 

348 if fail_rate > 5: 348 ↛ 349line 348 didn't jump to line 349 because the condition on line 348 was never true

349 return "warning" 

350 return "healthy" 

351 

352 

353def _trend_from_comparison(current_fail: float, previous_fail: float) -> str: 

354 if previous_fail <= 0: 354 ↛ 356line 354 didn't jump to line 356 because the condition on line 354 was always true

355 return "stable" 

356 diff: Final = current_fail - previous_fail 

357 if diff > 0.5: 

358 return "up" 

359 if diff < -0.5: 

360 return "down" 

361 return "stable" 

362 

363 

364def _aggregate_daily_metrics( 

365 metrics: "Sequence[_MetricsRowT]", id_of: "Callable[[_MetricsRowT], str]" 

366) -> Mapping[str, _MetricTotals]: 

367 agg: Final[dict[str, _MetricTotals]] = {} 

368 for m in metrics: 368 ↛ 369line 368 didn't jump to line 369 because the loop on line 368 never started

369 gid: str = id_of(m) 

370 if gid not in agg: 

371 agg[gid] = {"requests": 0, "passed": 0, "blocked": 0, "flagged": 0} 

372 agg[gid]["requests"] += int(m.requests_evaluated or 0) 

373 agg[gid]["passed"] += int(m.passed_count or 0) 

374 agg[gid]["blocked"] += int(m.blocked_count or 0) 

375 agg[gid]["flagged"] += int(m.flagged_count or 0) 

376 return agg 

377 

378 

379def _prev_fail_rates( 

380 metrics_prev: "Sequence[_MetricsRowT]", id_of: "Callable[[_MetricsRowT], str]" 

381) -> Mapping[str, float]: 

382 prev_agg_raw: Final[dict[str, _PrevPeriodCounts]] = {} 

383 for m in metrics_prev: 383 ↛ 384line 383 didn't jump to line 384 because the loop on line 383 never started

384 gid: str = id_of(m) 

385 r, b = int(m.requests_evaluated or 0), int(m.blocked_count or 0) 

386 if gid not in prev_agg_raw: 

387 prev_agg_raw[gid] = {"req": 0, "blocked": 0} 

388 prev_agg_raw[gid]["req"] += r 

389 prev_agg_raw[gid]["blocked"] += b 

390 return {gid: (100.0 * v["blocked"] / v["req"]) if v["req"] else 0.0 for gid, v in prev_agg_raw.items()} 

391 

392 

393def _chart_from_metrics(metrics: "Sequence[_DailyMetricsRow]") -> list[UsageChartPoint]: 

394 chart_by_date: Final[dict[str, _DailyPassBlocked]] = {} 

395 for m in metrics: 395 ↛ 396line 395 didn't jump to line 396 because the loop on line 395 never started

396 d = m.date 

397 if d not in chart_by_date: 

398 chart_by_date[d] = {"passed": 0, "blocked": 0} 

399 chart_by_date[d]["passed"] += int(m.passed_count or 0) 

400 chart_by_date[d]["blocked"] += int(m.blocked_count or 0) 

401 return [{"date": d, "passed": v["passed"], "blocked": v["blocked"]} for d, v in sorted(chart_by_date.items())] 

402 

403 

404_GuardrailStrField = Literal["guardrail_id", "guardrail_name"] 

405_GuardrailObjectField = Literal["litellm_params", "guardrail_info"] 

406 

407 

408@overload 

409def _get_guardrail_field(g: "_DbOrConfigGuardrail", field: _GuardrailStrField) -> str | None: ... 409 ↛ exitline 409 didn't return from function '_get_guardrail_field' because

410 

411 

412@overload 

413def _get_guardrail_field(g: "_DbOrConfigGuardrail", field: _GuardrailObjectField) -> object: ... 413 ↛ exitline 413 didn't return from function '_get_guardrail_field' because

414 

415 

416def _get_guardrail_field(g: "_DbOrConfigGuardrail", field: _GuardrailStrField | _GuardrailObjectField) -> object: 

417 """Read `field` off a guardrail whether it's a Prisma row (attr) or a dict/TypedDict (key).""" 

418 if isinstance(g, dict): 

419 return g.get(field) 

420 return getattr(g, field, None) 

421 

422 

423def _to_dict(value: object) -> dict[str, Any]: 

424 """Coerce a pydantic model (e.g. LitellmParams) / dict value into a plain dict.""" 

425 if isinstance(value, BaseModel): 

426 return value.model_dump(exclude_none=True) 

427 if isinstance(value, dict): 

428 return value 

429 return {} 

430 

431 

432def _field_str(mapping: Mapping[str, object], key: str, default: str) -> str: 

433 """Stringify `mapping[key]`, falling back to `default` when the key is absent.""" 

434 return str(mapping.get(key, default)) 

435 

436 

437def _get_guardrail_attrs(g: "_DbOrConfigGuardrail") -> tuple[str | None, str]: 

438 """Get (guardrail_id, display_name) from guardrail - handles Prisma model or dict.""" 

439 gid: Final = _get_guardrail_field(g, "guardrail_id") 

440 name: Final = _get_guardrail_field(g, "guardrail_name") 

441 return gid, (name or gid or "") 

442 

443 

444def _guardrail_overview_rows( 

445 guardrails: "Sequence[_DbOrConfigGuardrail]", 

446 agg: Mapping[str, _MetricTotals], 

447 prev_agg: Mapping[str, float], 

448 units_agg: Mapping[str, Mapping[str, int]], 

449 cost_agg: Mapping[str, float | None], 

450 untracked_agg: Mapping[str, Mapping[str, int]], 

451) -> list[UsageOverviewRow]: 

452 rows: Final[list[UsageOverviewRow]] = [] 

453 covered_keys: Final[set[str]] = set() 

454 for g in guardrails: 454 ↛ 455line 454 didn't jump to line 455 because the loop on line 454 never started

455 gid, display_name = _get_guardrail_attrs(g) 

456 # Metrics are keyed by logical name from spend log metadata; guardrails table uses UUID 

457 lookup_keys: Sequence[str] = [k for k in (display_name, gid) if k] 

458 covered_keys.update(lookup_keys) 

459 a: _MetricTotals = {"requests": 0, "passed": 0, "blocked": 0, "flagged": 0} 

460 for k in lookup_keys: 

461 if k in agg: 

462 a = agg[k] 

463 break 

464 req, blocked = a["requests"], a["blocked"] 

465 fail_rate = (100.0 * blocked / req) if req else 0.0 

466 litellm_params = _to_dict(_get_guardrail_field(g, "litellm_params")) 

467 provider = _field_str(litellm_params, "guardrail", "Unknown") 

468 guardrail_info = _to_dict(_get_guardrail_field(g, "guardrail_info")) 

469 gtype = _field_str(guardrail_info, "type", "Guardrail") 

470 prev_fail = 0.0 

471 for k in lookup_keys: 

472 if k in prev_agg: 

473 prev_fail = float(prev_agg.get(k, 0.0) or 0.0) 

474 break 

475 trend = _trend_from_comparison(fail_rate, prev_fail) 

476 rows.append( 

477 UsageOverviewRow( 

478 id=gid, 

479 name=display_name or str(gid), 

480 type=gtype, 

481 provider=provider, 

482 requestsEvaluated=req, 

483 failRate=round(fail_rate, 1), 

484 avgScore=None, 

485 avgLatency=None, 

486 status=_status_from_fail_rate(fail_rate), 

487 trend=trend, 

488 usageUnits=_first_match(lookup_keys, units_agg, _EMPTY_UNITS), 

489 cost=_first_match(lookup_keys, cost_agg, None), 

490 untrackedUsageUnits=_first_match(lookup_keys, untracked_agg, _EMPTY_UNITS), 

491 ) 

492 ) 

493 # Add rows for guardrails with metrics but not in guardrails table (e.g. MCP, config) 

494 for agg_key, a in agg.items(): 494 ↛ 495line 494 didn't jump to line 495 because the loop on line 494 never started

495 if agg_key in covered_keys or a["requests"] == 0: 

496 continue 

497 req, blocked = a["requests"], a["blocked"] 

498 fail_rate = (100.0 * blocked / req) if req else 0.0 

499 prev_fail = float(prev_agg.get(agg_key, 0.0) or 0.0) 

500 trend = _trend_from_comparison(fail_rate, prev_fail) 

501 rows.append( 

502 UsageOverviewRow( 

503 id=agg_key, 

504 name=agg_key, 

505 type="Guardrail", 

506 provider="Custom", 

507 requestsEvaluated=req, 

508 failRate=round(fail_rate, 1), 

509 avgScore=None, 

510 avgLatency=None, 

511 status=_status_from_fail_rate(fail_rate), 

512 trend=trend, 

513 usageUnits=units_agg.get(agg_key, _EMPTY_UNITS), 

514 cost=cost_agg.get(agg_key), 

515 untrackedUsageUnits=untracked_agg.get(agg_key, _EMPTY_UNITS), 

516 ) 

517 ) 

518 return rows 

519 

520 

521def _policy_overview_rows( 

522 policies: "Sequence[prisma_models.LiteLLM_PolicyTable]", 

523 agg: Mapping[str, _MetricTotals], 

524 prev_agg: Mapping[str, float], 

525) -> list[UsageOverviewRow]: 

526 rows: Final[list[UsageOverviewRow]] = [] 

527 for p in policies: 

528 pid = p.policy_id 

529 a = agg.get(pid, {"requests": 0, "passed": 0, "blocked": 0, "flagged": 0}) 

530 req, blocked = a["requests"], a["blocked"] 

531 fail_rate = (100.0 * blocked / req) if req else 0.0 

532 trend = _trend_from_comparison(fail_rate, prev_agg.get(pid, 0.0)) 

533 rows.append( 

534 UsageOverviewRow( 

535 id=pid, 

536 name=p.policy_name or pid, 

537 type="Policy", 

538 provider="LiteLLM", 

539 requestsEvaluated=req, 

540 failRate=round(fail_rate, 1), 

541 avgScore=None, 

542 avgLatency=None, 

543 status=_status_from_fail_rate(fail_rate), 

544 trend=trend, 

545 usageUnits=_EMPTY_UNITS, 

546 cost=None, 

547 untrackedUsageUnits=_EMPTY_UNITS, 

548 ) 

549 ) 

550 return rows 

551 

552 

553@router.get( 

554 "/guardrails/usage/overview", 

555 tags=["Guardrails"], 

556 dependencies=[Depends(user_api_key_auth)], 

557 response_model=UsageOverviewResponse, 

558) 

559async def guardrails_usage_overview( 

560 start_date: str | None = Query(None, description="YYYY-MM-DD"), 

561 end_date: str | None = Query(None, description="YYYY-MM-DD"), 

562 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), 

563): 

564 """Return guardrail performance overview for the dashboard.""" 

565 from litellm.proxy.proxy_server import prisma_client 

566 

567 if prisma_client is None: 567 ↛ 568line 567 didn't jump to line 568 because the condition on line 567 was never true

568 return _EMPTY_OVERVIEW 

569 

570 start, end = _resolve_usage_window(start_date, end_date) 

571 

572 from litellm.proxy.guardrails.guardrail_registry import IN_MEMORY_GUARDRAIL_HANDLER 

573 

574 try: 

575 db_guardrails: Final = await _guardrails_table(prisma_client).find_many() 

576 seen_ids: Final = {gid for g in db_guardrails if (gid := _get_guardrail_field(g, "guardrail_id")) is not None} 

577 config_guardrails: Final = [ 

578 g for g in IN_MEMORY_GUARDRAIL_HANDLER.list_config_guardrails() if g.get("guardrail_id") not in seen_ids 

579 ] 

580 guardrails: Final[Sequence[_DbOrConfigGuardrail]] = [*db_guardrails, *config_guardrails] 

581 

582 # Daily metrics in range 

583 metrics: Final[Sequence[prisma_models.LiteLLM_DailyGuardrailMetrics]] = await _find_daily_guardrail_metrics( 

584 prisma_client, where={"date": {"gte": start, "lte": end}} 

585 ) 

586 

587 # Previous period for trend 

588 start_prev: Final = (date.fromisoformat(start) - timedelta(days=7)).isoformat() 

589 metrics_prev: Sequence[prisma_models.LiteLLM_DailyGuardrailMetrics] = await _find_daily_guardrail_metrics( 

590 prisma_client, where={"date": {"gte": start_prev, "lt": start}} 

591 ) 

592 

593 units_where: Final[prisma_types.LiteLLM_DailyGuardrailUsageUnitsWhereInput] = { 

594 "date": {"gte": start, "lte": end} 

595 } 

596 units_rows: Final[ 

597 Sequence[prisma_models.LiteLLM_DailyGuardrailUsageUnits] 

598 ] = await _find_daily_guardrail_usage_units(prisma_client, where=units_where) 

599 

600 agg: Final = _aggregate_daily_metrics(metrics, lambda m: m.guardrail_id) 

601 prev_agg: Final = _prev_fail_rates(metrics_prev, lambda m: m.guardrail_id) 

602 units_agg: Final = _by(units_rows, lambda r: r.guardrail_id, _sum_counter_units) 

603 cost_agg: Final = _by(units_rows, lambda r: r.guardrail_id, _sum_tracked_cost) 

604 untracked_agg: Final = _by(units_rows, lambda r: r.guardrail_id, _sum_untracked_units) 

605 chart: Final = _chart_from_metrics(metrics) 

606 total_requests: Final = sum(a["requests"] for a in agg.values()) 

607 total_blocked: Final = sum(a["blocked"] for a in agg.values()) 

608 pass_rate: Final = (100.0 * (total_requests - total_blocked) / total_requests) if total_requests else 100.0 

609 rows: Final = _guardrail_overview_rows(guardrails, agg, prev_agg, units_agg, cost_agg, untracked_agg) 

610 return UsageOverviewResponse( 

611 rows=rows, 

612 chart=chart, 

613 totalRequests=total_requests, 

614 totalBlocked=total_blocked, 

615 passRate=round(pass_rate, 1), 

616 totalUsageUnits=_sum_counter_units(units_rows), 

617 totalCost=_sum_tracked_cost(units_rows), 

618 totalUntrackedUsageUnits=_sum_untracked_units(units_rows), 

619 ) 

620 except Exception as e: 

621 from litellm.proxy.utils import handle_exception_on_proxy 

622 

623 raise handle_exception_on_proxy(e) 

624 

625 

626@router.get( 

627 "/guardrails/usage/detail/{guardrail_id}", 

628 tags=["Guardrails"], 

629 dependencies=[Depends(user_api_key_auth)], 

630 response_model=UsageDetailResponse, 

631) 

632async def guardrails_usage_detail( 

633 guardrail_id: str, 

634 start_date: str | None = Query(None), 

635 end_date: str | None = Query(None), 

636 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), 

637): 

638 """Return single guardrail usage metrics and time series.""" 

639 from litellm.proxy.proxy_server import prisma_client 

640 

641 if prisma_client is None: 641 ↛ 642line 641 didn't jump to line 642 because the condition on line 641 was never true

642 from fastapi import HTTPException 

643 

644 raise HTTPException(status_code=500, detail="Prisma client not initialized") 

645 

646 start, end = _resolve_usage_window(start_date, end_date) 

647 

648 from litellm.proxy.guardrails.guardrail_registry import IN_MEMORY_GUARDRAIL_HANDLER 

649 

650 guardrail = await _guardrails_table(prisma_client).find_unique(where={"guardrail_id": guardrail_id}) 

651 if guardrail is None: 651 ↛ 653line 651 didn't jump to line 653 because the condition on line 651 was always true

652 guardrail = IN_MEMORY_GUARDRAIL_HANDLER.get_config_guardrail_by_id(guardrail_id=guardrail_id) 

653 if guardrail is None: 653 ↛ 659line 653 didn't jump to line 659 because the condition on line 653 was always true

654 from fastapi import HTTPException 

655 

656 raise HTTPException(status_code=404, detail="Guardrail not found") 

657 

658 # Metrics are keyed by logical name (from spend log metadata), not UUID 

659 logical_id: Final = _get_guardrail_field(guardrail, "guardrail_name") 

660 metric_ids: Final = [i for i in (logical_id, guardrail_id) if i] 

661 

662 metrics: Final[Sequence[prisma_models.LiteLLM_DailyGuardrailMetrics]] = await _find_daily_guardrail_metrics( 

663 prisma_client, 

664 where={ 

665 "guardrail_id": {"in": metric_ids}, 

666 "date": {"gte": start, "lte": end}, 

667 }, 

668 ) 

669 start_prev: Final = (date.fromisoformat(start) - timedelta(days=7)).isoformat() 

670 metrics_prev: Final[Sequence[prisma_models.LiteLLM_DailyGuardrailMetrics]] = await _find_daily_guardrail_metrics( 

671 prisma_client, 

672 where={ 

673 "guardrail_id": {"in": metric_ids}, 

674 "date": {"gte": start_prev, "lt": start}, 

675 }, 

676 ) 

677 units_where: Final[prisma_types.LiteLLM_DailyGuardrailUsageUnitsWhereInput] = { 

678 "guardrail_id": {"in": metric_ids}, 

679 "date": {"gte": start, "lte": end}, 

680 } 

681 units_rows: Final[ 

682 Sequence[prisma_models.LiteLLM_DailyGuardrailUsageUnits] 

683 ] = await _find_daily_guardrail_usage_units(prisma_client, where=units_where) 

684 

685 requests: Final = sum(int(m.requests_evaluated or 0) for m in metrics) 

686 blocked: Final = sum(int(m.blocked_count or 0) for m in metrics) 

687 fail_rate: Final = (100.0 * blocked / requests) if requests else 0.0 

688 

689 prev_blocked: Final = sum(int(m.blocked_count or 0) for m in metrics_prev) 

690 prev_req: Final = sum(int(m.requests_evaluated or 0) for m in metrics_prev) 

691 prev_fail: Final = (100.0 * prev_blocked / prev_req) if prev_req else 0.0 

692 trend: Final = _trend_from_comparison(fail_rate, prev_fail) 

693 

694 # Aggregate by date in case metrics exist under both UUID and logical name 

695 ts_by_date: Final[dict[str, _DailyPassBlocked]] = {} 

696 for m in metrics: 

697 d = m.date 

698 if d not in ts_by_date: 

699 ts_by_date[d] = {"passed": 0, "blocked": 0} 

700 ts_by_date[d]["passed"] += int(m.passed_count or 0) 

701 ts_by_date[d]["blocked"] += int(m.blocked_count or 0) 

702 time_series: Final[list[UsageChartPoint]] = [ 

703 {"date": d, "passed": v["passed"], "blocked": v["blocked"], "score": None} 

704 for d, v in sorted(ts_by_date.items()) 

705 ] 

706 litellm_params: Final = _to_dict(_get_guardrail_field(guardrail, "litellm_params")) 

707 guardrail_info: Final = _to_dict(_get_guardrail_field(guardrail, "guardrail_info")) 

708 _guardrail_name: Final = _get_guardrail_field(guardrail, "guardrail_name") 

709 daily_unit_sums: Final = sorted(_by(units_rows, lambda r: r.date, _sum_counter_units).items()) 

710 daily_cost: Final = _by(units_rows, lambda r: r.date, _sum_tracked_cost) 

711 units_daily: Final = tuple( 

712 UsageUnitsDailyPoint(date=d, units=units, cost=daily_cost.get(d)) for d, units in daily_unit_sums 

713 ) 

714 

715 return UsageDetailResponse(