test: validate daily alerts and document the public API

This commit is contained in:
moe-berri 2026-11-05 11:49:51 +08:00 • committed by ishaan-j-berri
parent f82c79a6f0
commit c741d1271d
10 changed files with 958 additions and 94 deletions

View file

@ -537,6 +537,7 @@ jobs:
tests/unit/proxy/management_endpoints
tests/unit/proxy/management_helpers
tests/unit/proxy/list_api
tests/unit/proxy/alerting_endpoints
tests/unit/proxy/analytics_endpoints
tests/unit/proxy/decisions_endpoints
tests/unit/proxy/memory

View file

@ -0,0 +1,71 @@
# Daily usage alerts
Daily usage alerts compare a completed UTC day with earlier recorded days. They cover spend, request count, average latency, and failed-request count for the gateway, a key, a user, a team, or a model. Open an incident to inspect its saved comparison and follow its Activity link to the corresponding usage view.
This feature requires a database. Apply the normal LiteLLM database migrations when deploying a version that includes usage alerts. The API and dashboard use the same persisted rules and incidents; no enterprise license is required.
## Read incidents
Set `LITELLM_API_BASE` to your gateway's base URL and `LITELLM_MANAGEMENT_KEY` to a management-capable credential. Keep the key in your environment.
```bash
curl "$LITELLM_API_BASE/usage/alerts?status=open&limit=50" \
-H "Authorization: Bearer $LITELLM_MANAGEMENT_KEY"
```
The response has `data`, `total`, `limit`, and `offset`. Each incident includes its scope, metric, direction, status, UTC day, observed value, baseline value, threshold, baseline date range, and `activity_path`. The Activity path is relative to the dashboard mount: for example, `/api-keys?...` is opened below the dashboard's `/ui` mount.
`GET /usage/alerts/{alert_id}` returns one incident. List filters are `scope_type`, `scope_id`, `metric`, `status`, `start_date`, and `end_date`. Dates use `YYYY-MM-DD`; date filters include both endpoints. The maximum page size is 100.
## Create a rule
This example alerts when a team's daily spend is at least three times its baseline and the absolute change is at least $10. Replace `team-example` with the team's ID.
```bash
curl "$LITELLM_API_BASE/usage/alert-rules" \
-H "Authorization: Bearer $LITELLM_MANAGEMENT_KEY" \
-H "Content-Type: application/json" \
-d '{
"name": "Team daily spend",
"scope_type": "team",
"scope_id": "team-example",
"metric": "spend",
"direction": "increase",
"multiplier": 3,
"baseline_days": 7,
"min_history_days": 3,
"min_requests": 100,
"min_value": 10,
"enabled": true
}'
```
The baseline is the mean across recorded days in the preceding window. Missing reference days are unknown. At least `min_history_days` recorded days are required. A decrease uses `baseline / multiplier`; a multiplier of 2 detects a value at or below half the baseline. `min_value` is the minimum absolute difference in the metric's own units.
| Metric | Unit |
| --- | --- |
| `spend` | USD |
| `requests` | Request count |
| `latency` | Average response time in milliseconds |
| `errors` | Failed-request count |
The evaluator runs against daily aggregates. An hourly scheduler does not make these rolling hourly alerts. A complete UTC day and sufficient ingestion coverage are required before detecting drops; a missing flush must not become a false traffic-drop incident. Rule responses expose `last_evaluated_day` and `evaluation_status` so you can distinguish a completed evaluation from insufficient history or coverage.
Use `scope_type: "deployment"` with no `scope_id` for the gateway. For `model`, use the underlying model identity recorded in Activity. For `key`, use its reporting ID, never the secret API key. Omitting `scope_id` for key, user, team, or model evaluates all entities of that type and requires a gateway administrator.
## Manage rules and incidents
- `GET /usage/alert-rules` lists rules; `GET /usage/alert-rules/{rule_id}` reads one.
- `PUT /usage/alert-rules/{rule_id}` replaces the rule configuration. Send the configuration fields from the create example, including any notification settings you want to retain. Set `enabled` to `false` to pause it.
- `DELETE /usage/alert-rules/{rule_id}` removes a rule from evaluation and preserves existing incidents. Deleted defaults stay deleted.
- `PATCH /usage/alerts/{alert_id}` with `{"status":"dismissed"}` dismisses an incident. `open` and `resolved` are also accepted.
Consecutive breached days update one incident. Re-evaluating the same day does not reopen a dismissed or resolved incident. After recovery, a later breach starts a new incident. Editing a rule does not rewrite historical comparisons.
Rules can specify `notification_channels: ["slack", "email"]` and `email_recipients: ["alerts@example.com"]`. Notification transports must already be configured on the gateway. An empty channel list keeps incidents available through the API and dashboard. Successful delivery is recorded per channel so a failed channel can retry independently.
## Access
Gateway administrators can manage every scope. Admin viewers can read every scope. Internal users can manage their own user and key scopes; team administrators can manage their team's scope and keys. Team members can read their team's incidents. Read-only users cannot change rules or incident state, and inference-only keys cannot access these endpoints.
Non-admin list requests must specify a scope they can access, for example `?scope_type=user&scope_id=user-example`. A `403` indicates that the credential or scope lacks access. A `503` indicates that the gateway has no database connection.

View file

@ -579,6 +579,10 @@ class LiteLLMRoutes(enum.Enum):
)
llm_api_routes_without_mcp = sorted(set(llm_api_routes) - set(mcp_inference_routes))
info_routes = [
"/usage/alert-rules",
"/usage/alert-rules/{rule_id}",
"/usage/alerts",
"/usage/alerts/{alert_id}",
"/key/info",
"/key/health",
"/team/info",

View file

@ -1,7 +1,7 @@
"""Public management API for daily usage alert rules and incidents."""
from datetime import date
from typing import Annotated, Final
from typing import Annotated, Final, TypeAlias
from fastapi import APIRouter, Depends, HTTPException, Query, Response
@ -32,11 +32,11 @@ def usage_alert_repository(database: DatabaseDep) -> UsageAlertsRepository:
return UsageAlertsRepository.from_prisma(database.prisma)
User = Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)]
Repository = Annotated[UsageAlertsRepository, Depends(usage_alert_repository)]
Limit = Annotated[int, Query(ge=1, le=100)]
Offset = Annotated[int, Query(ge=0, le=100000)]
ScopeId = Annotated[str | None, Query(min_length=1, max_length=1024)]
User: TypeAlias = Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)]
Repository: TypeAlias = Annotated[UsageAlertsRepository, Depends(usage_alert_repository)]
Limit: TypeAlias = Annotated[int, Query(ge=1, le=100)]
Offset: TypeAlias = Annotated[int, Query(ge=0, le=100000)]
ScopeId: TypeAlias = Annotated[str | None, Query(min_length=1, max_length=1024)]
async def require_scope(
@ -101,7 +101,10 @@ async def list_usage_alert_rules(
@router.post("/alert-rules", response_model=UsageAlertRule, status_code=201)
async def create_usage_alert_rule(data: UsageAlertRuleInput, user: User, repository: Repository) -> UsageAlertRule:
await require_scope(user, repository, data.scope_type, data.scope_id, write=True)
return await repository.save_rule(data, user.user_id)
saved: Final = await repository.save_rule(data, user.user_id)
if saved is None:
raise HTTPException(status_code=409, detail="Alert rule could not be saved")
return saved
@router.get("/alert-rules/{rule_id}", response_model=UsageAlertRule)
@ -122,7 +125,10 @@ async def update_usage_alert_rule(
raise HTTPException(status_code=404, detail="Alert rule not found")
await require_scope(user, repository, rule.scope_type, rule.scope_id, write=True)
await require_scope(user, repository, data.scope_type, data.scope_id, write=True)
return await repository.save_rule(data, user.user_id, existing=rule)
saved: Final = await repository.save_rule(data, user.user_id, existing=rule)
if saved is None:
raise HTTPException(status_code=404, detail="Alert rule not found")
return saved
@router.delete("/alert-rules/{rule_id}", status_code=204)

View file

@ -1,8 +1,14 @@
"""Bounded, writer-consistent storage for daily alert rules and episodes."""
import json
from collections.abc import Mapping
from datetime import date, datetime, timedelta, timezone
from typing import TYPE_CHECKING, Final, Protocol, cast
from typing import ( # PrismaWrapper forwards generated client methods dynamically
TYPE_CHECKING,
Final,
Protocol,
cast, # noqa: TID251 # PrismaWrapper forwards generated client methods dynamically
)
from urllib.parse import urlencode
from uuid import NAMESPACE_URL, uuid4, uuid5
@ -31,25 +37,20 @@ class AlertDatabase(Protocol):
async def query_raw(self, query: str, *args: object) -> object: ...
class _Record(BaseModel):
record: dict[str, object]
class _Record(BaseModel, frozen=True):
record: Mapping[str, object]
class _Count(BaseModel):
class _Count(BaseModel, frozen=True):
count: int
class _Episode(_Record):
evaluation_day: date
breached: bool
class _Pending(_Record):
rule: dict[str, object]
class _Pending(_Record, frozen=True):
rule: Mapping[str, object]
notified_channels: tuple[UsageAlertChannel, ...]
class _Ownership(BaseModel):
class _Ownership(BaseModel, frozen=True):
user_id: str | None
team_id: str | None
@ -71,7 +72,7 @@ def activity_path(
+ "?"
+ urlencode(
{
"range": "custom",
"tab": "keys",
"from": (day - timedelta(days=baseline_days)).isoformat(),
"to": day.isoformat(),
"alert": alert_id,
@ -89,13 +90,47 @@ def activity_path(
return route + "?" + urlencode(query)
def _observation_payload(rule: UsageAlertRule, item: UsageAlertObservation, now: datetime) -> Mapping[str, object]:
alert_id: Final = str(
uuid5(NAMESPACE_URL, f"litellm:usage-alert:{rule.id}:{item.entity_id or ''}:{item.day.isoformat()}")
)
incident: Final = UsageAlert(
id=alert_id,
rule_id=rule.id,
rule_name=rule.name,
scope_type=rule.scope_type,
scope_id=item.entity_id,
metric=rule.metric,
direction=rule.direction,
status="open",
day=item.day,
baseline_start=item.day - timedelta(days=rule.baseline_days),
baseline_end=item.day - timedelta(days=1),
baseline_sample_days=item.baseline_sample_days,
observed_value=item.observed,
baseline_value=item.baseline,
threshold=item.threshold,
first_seen=now,
last_seen=now,
activity_path=activity_path(rule.scope_type, item.entity_id, item.day, alert_id, rule.baseline_days),
)
return {
"scope_key": item.entity_id or "",
"breached": item.breached,
"day": item.day.isoformat(),
"record": incident.model_dump(mode="json"),
}
class UsageAlertsRepository:
def __init__(self, database: AlertDatabase):
def __init__(self, database: AlertDatabase) -> None:
self.database: Final = database
@classmethod
def from_prisma(cls, prisma: "PrismaClient") -> "UsageAlertsRepository":
return cls(cast(AlertDatabase, prisma.writer_db))
return cls(
cast(AlertDatabase, prisma.writer_db) # cast-ok: PrismaWrapper forwards query_raw to its generated client
)
async def get_rule(self, rule_id: str) -> UsageAlertRule | None:
rows: Final = TypeAdapter(tuple[_Record, ...]).validate_python(
@ -107,14 +142,16 @@ class UsageAlertsRepository:
async def save_rule(
self, data: UsageAlertRuleInput, created_by: str | None, existing: UsageAlertRule | None = None
) -> UsageAlertRule:
) -> UsageAlertRule | None:
now: Final = datetime.now(timezone.utc)
rule: Final = UsageAlertRule(
**data.model_dump(),
id=existing.id if existing else str(uuid4()),
created_by=existing.created_by if existing else created_by,
created_at=existing.created_at if existing else now,
updated_at=now,
rule: Final = UsageAlertRule.model_validate(
{
**data.model_dump(),
"id": existing.id if existing else str(uuid4()),
"created_by": existing.created_by if existing else created_by,
"created_at": existing.created_at if existing else now,
"updated_at": now,
}
)
rows: Final = await self.database.query_raw(
"""INSERT INTO "LiteLLM_UsageAlertRule" (id,scope_type,scope_id,enabled,record) VALUES ($1,$2,$3,$4,$5::jsonb)
@ -126,7 +163,8 @@ class UsageAlertsRepository:
rule.enabled,
rule.model_dump_json(),
)
return UsageAlertRule.model_validate(TypeAdapter(tuple[_Record, ...]).validate_python(rows)[0].record)
saved: Final = TypeAdapter(tuple[_Record, ...]).validate_python(rows)
return UsageAlertRule.model_validate(saved[0].record) if saved else None
async def delete_rule(self, rule_id: str) -> None:
await self.database.query_raw(
@ -269,24 +307,27 @@ class UsageAlertsRepository:
async def ensure_default_rules(self, rules: tuple[UsageAlertRuleInput, ...]) -> None:
for rule in rules:
rule_id: Final = str(
uuid5(NAMESPACE_URL, f"litellm:usage-alert:{rule.scope_type}:{rule.metric}:{rule.direction}")
)
now: Final = datetime.now(timezone.utc)
record: Final = UsageAlertRule(
**rule.model_dump(), id=rule_id, created_at=now, updated_at=now, created_by=None
)
await self.database.query_raw(
'INSERT INTO "LiteLLM_UsageAlertRule" (id,scope_type,scope_id,enabled,record) VALUES ($1,$2,$3,$4,$5::jsonb) ON CONFLICT(id) DO NOTHING RETURNING id',
rule_id,
rule.scope_type,
rule.scope_id,
rule.enabled,
record.model_dump_json(),
)
await self._ensure_default_rule(rule)
async def _ensure_default_rule(self, rule: UsageAlertRuleInput) -> None:
rule_id: Final = str(
uuid5(NAMESPACE_URL, f"litellm:usage-alert:{rule.scope_type}:{rule.metric}:{rule.direction}")
)
now: Final = datetime.now(timezone.utc)
record: Final = UsageAlertRule.model_validate(
{**rule.model_dump(), "id": rule_id, "created_at": now, "updated_at": now, "created_by": None}
)
await self.database.query_raw(
'INSERT INTO "LiteLLM_UsageAlertRule" (id,scope_type,scope_id,enabled,record) VALUES ($1,$2,$3,$4,$5::jsonb) ON CONFLICT(id) DO NOTHING RETURNING id',
rule_id,
rule.scope_type,
rule.scope_id,
rule.enabled,
record.model_dump_json(),
)
async def list_breached_entity_ids(self, rule_id: str, limit: int = 500, offset: int = 0) -> tuple[str | None, ...]:
rows: Final = TypeAdapter(tuple[dict[str, str | None], ...]).validate_python(
rows: Final = TypeAdapter(tuple[Mapping[str, str | None], ...]).validate_python(
await self.database.query_raw(
'SELECT DISTINCT scope_id FROM "LiteLLM_UsageAlertIncident" WHERE rule_id=$1 AND breached ORDER BY scope_id NULLS FIRST LIMIT $2 OFFSET $3',
rule_id,
@ -300,11 +341,11 @@ class UsageAlertsRepository:
self, rule: UsageAlertRule, day: date, status: UsageAlertEvaluationStatus
) -> None:
await self.database.query_raw(
"""UPDATE "LiteLLM_UsageAlertRule" SET record=record || jsonb_build_object('last_evaluated_day',$2::text,'evaluation_status',$3::text) WHERE id=$1 AND record->>'updated_at'=$4 AND NOT deleted RETURNING id""",
"""UPDATE "LiteLLM_UsageAlertRule" SET record=record || jsonb_build_object('last_evaluated_day',$2::text,'evaluation_status',$3::text) WHERE id=$1 AND (record->>'updated_at')::timestamptz=$4::timestamptz AND NOT deleted RETURNING id""",
rule.id,
day.isoformat(),
status,
rule.model_dump(mode="json")["updated_at"],
rule.updated_at.isoformat(),
)
async def record_observation(
@ -340,38 +381,7 @@ class UsageAlertsRepository:
if len(observations) > 500 or len({item.entity_id for item in observations}) != len(observations):
raise ValueError("An observation batch must contain at most 500 distinct entities")
now: Final = datetime.now(timezone.utc)
records: Final = []
for item in observations:
alert_id: Final = str(
uuid5(NAMESPACE_URL, f"litellm:usage-alert:{rule.id}:{item.entity_id or ''}:{item.day.isoformat()}")
)
incident: Final = UsageAlert(
id=alert_id,
rule_id=rule.id,
rule_name=rule.name,
scope_type=rule.scope_type,
scope_id=item.entity_id,
metric=rule.metric,
direction=rule.direction,
status="open",
day=item.day,
baseline_start=item.day - timedelta(days=rule.baseline_days),
baseline_end=item.day - timedelta(days=1),
observed_value=item.observed,
baseline_value=item.baseline,
threshold=item.threshold,
first_seen=now,
last_seen=now,
activity_path=activity_path(rule.scope_type, item.entity_id, item.day, alert_id, rule.baseline_days),
)
records.append(
{
"scope_key": item.entity_id or "",
"breached": item.breached,
"day": item.day.isoformat(),
"record": incident.model_dump(mode="json"),
}
)
records: Final = tuple(_observation_payload(rule, item, now) for item in observations)
rows: Final = TypeAdapter(tuple[_Record, ...]).validate_python(
await self.database.query_raw(
"""
@ -381,7 +391,7 @@ class UsageAlertsRepository:
FROM input x LEFT JOIN LATERAL (
SELECT * FROM "LiteLLM_UsageAlertIncident" WHERE rule_id=$1 AND scope_key=x.scope_key ORDER BY evaluation_day DESC,id DESC LIMIT 1
) p ON true
WHERE EXISTS (SELECT 1 FROM "LiteLLM_UsageAlertRule" WHERE id=$1 AND enabled AND NOT deleted AND record->>'updated_at'=$3)
WHERE EXISTS (SELECT 1 FROM "LiteLLM_UsageAlertRule" WHERE id=$1 AND enabled AND NOT deleted AND (record->>'updated_at')::timestamptz=$3::timestamptz)
), updated AS (
UPDATE "LiteLLM_UsageAlertIncident" i SET
evaluation_day=p.day, breached=p.breached,
@ -403,7 +413,7 @@ class UsageAlertsRepository:
""",
rule.id,
json.dumps(records),
rule.model_dump(mode="json")["updated_at"],
rule.updated_at.isoformat(),
)
)
return tuple(UsageAlert.model_validate(row.record) for row in rows)
@ -450,7 +460,7 @@ class UsageAlertsRepository:
)
async def team_role(self, team_id: str, user_id: str) -> str | None:
rows: Final = TypeAdapter(tuple[dict[str, str], ...]).validate_python(
rows: Final = TypeAdapter(tuple[Mapping[str, str], ...]).validate_python(
await self.database.query_raw(
"""SELECT member->>'role' AS role FROM "LiteLLM_TeamTable", jsonb_array_elements(members_with_roles) member WHERE team_id=$1 AND member->>'user_id'=$2 LIMIT 1""",
team_id,

View file

@ -1,16 +1,16 @@
"""Daily usage alert rules and incidents shared by the API and evaluator."""
from datetime import date, datetime
from typing import Literal, Self
from typing import Literal, Self, TypeAlias
from pydantic import BaseModel, ConfigDict, Field, model_validator
UsageAlertScope = Literal["deployment", "key", "user", "team", "model"]
UsageAlertMetric = Literal["spend", "requests", "latency", "errors"]
UsageAlertDirection = Literal["increase", "decrease"]
UsageAlertStatus = Literal["open", "resolved", "dismissed"]
UsageAlertChannel = Literal["email", "slack"]
UsageAlertEvaluationStatus = Literal["pending", "complete", "insufficient_coverage", "insufficient_history"]
UsageAlertScope: TypeAlias = Literal["deployment", "key", "user", "team", "model"]
UsageAlertMetric: TypeAlias = Literal["spend", "requests", "latency", "errors"]
UsageAlertDirection: TypeAlias = Literal["increase", "decrease"]
UsageAlertStatus: TypeAlias = Literal["open", "resolved", "dismissed"]
UsageAlertChannel: TypeAlias = Literal["email", "slack"]
UsageAlertEvaluationStatus: TypeAlias = Literal["pending", "complete", "insufficient_coverage", "insufficient_history"]
class UsageAlertRuleInput(BaseModel):
@ -36,7 +36,7 @@ class UsageAlertRuleInput(BaseModel):
raise ValueError("min_history_days must not exceed baseline_days")
if self.scope_type == "deployment" and self.scope_id is not None:
raise ValueError("Deployment alerts cover the whole gateway; omit scope_id")
if self.scope_type == "key" and self.scope_id is not None and self.scope_id.startswith("sk-"):
if self.scope_type == "key" and self.scope_id is not None and self.scope_id.removeprefix("raw:").startswith("sk-"):
raise ValueError("Use the key's reporting ID, never the secret key")
if len(set(self.notification_channels)) != len(self.notification_channels):
raise ValueError("notification_channels must be unique")
@ -72,6 +72,7 @@ class UsageAlert(BaseModel):
day: date
baseline_start: date
baseline_end: date
baseline_sample_days: int | None = Field(default=None, ge=1, le=30)
observed_value: float
baseline_value: float
threshold: float
@ -85,14 +86,14 @@ class UsageAlertUpdate(BaseModel):
status: UsageAlertStatus
class UsageAlertRuleList(BaseModel):
class UsageAlertRuleList(BaseModel, frozen=True):
data: tuple[UsageAlertRule, ...]
total: int
limit: int
offset: int
class UsageAlertList(BaseModel):
class UsageAlertList(BaseModel, frozen=True):
data: tuple[UsageAlert, ...]
total: int
limit: int
@ -112,5 +113,6 @@ class UsageAlertObservation(BaseModel):
day: date
observed: float
baseline: float
baseline_sample_days: int | None = Field(default=None, ge=1, le=30)
threshold: float
breached: bool

View file

@ -0,0 +1,78 @@
from hashlib import sha256
from typing import Final
from integration._support.client import Gateway, string_value
def test_scoped_rules_use_real_management_auth_and_preserve_read_only_roles(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
owner: Final = scenario.user(user_role="internal_user")
outsider: Final = scenario.user(user_role="internal_user")
viewer: Final = scenario.user(user_role="proxy_admin_viewer")
owner_key: Final = scenario.key(user_id=owner, key_type="management")
outsider_key: Final = scenario.key(user_id=outsider, key_type="management")
viewer_key: Final = scenario.key(user_id=viewer, key_type="read_only")
inference_key: Final = scenario.key(user_id=owner, key_type="llm_api")
config: Final = {
"name": "Daily key spend",
"scope_type": "key",
"scope_id": sha256(inference_key.encode()).hexdigest(),
"metric": "spend",
"direction": "increase",
"multiplier": 3,
}
created: Final = gateway.request("POST", "/usage/alert-rules", config, key=owner_key)
assert created.status_code == 201, created.text
rule_id: Final = string_value(created.json()["id"])
path: Final = f"/usage/alert-rules/{rule_id}"
try:
assert gateway.request("GET", path, key=owner_key).status_code == 200
assert gateway.request("GET", path, key=outsider_key).status_code == 403
assert gateway.request("GET", path, key=inference_key).status_code == 403
assert gateway.request("GET", path, key=viewer_key).status_code == 200
assert gateway.request("PUT", path, config, key=viewer_key).status_code == 403
assert gateway.request("PUT", path, config, key=outsider_key).status_code == 403
assert (
gateway.request("POST", "/usage/alert-rules", {**config, "scope_id": None}, key=owner_key).status_code
== 403
)
disabled: Final = gateway.request("PUT", path, {**config, "enabled": False}, key=owner_key)
assert disabled.status_code == 200 and disabled.json()["enabled"] is False
page: Final = gateway.request(
"GET",
"/usage/alert-rules",
key=owner_key,
params={"scope_type": "key", "scope_id": str(config["scope_id"]), "limit": "1"},
)
assert page.status_code == 200 and page.json()["total"] == 1
assert page.json()["data"][0]["id"] == rule_id
finally:
assert gateway.request("DELETE", path, key=owner_key).status_code == 204
assert gateway.request("GET", path, key=owner_key).status_code == 404
def test_team_admin_can_manage_team_alerts_but_member_can_only_read(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
team_id: Final = scenario.team()
admin_id: Final = scenario.member(team_id, role="admin")
member_id: Final = scenario.member(team_id, role="user")
admin_key: Final = scenario.key(user_id=admin_id, key_type="management")
member_key: Final = scenario.key(user_id=member_id, key_type="management")
config: Final = {
"name": "Daily team requests",
"scope_type": "team",
"scope_id": team_id,
"metric": "requests",
"direction": "decrease",
"multiplier": 2,
}
created: Final = gateway.request("POST", "/usage/alert-rules", config, key=admin_key)
assert created.status_code == 201, created.text
rule_id: Final = string_value(created.json()["id"])
path: Final = f"/usage/alert-rules/{rule_id}"
try:
assert gateway.request("GET", path, key=member_key).status_code == 200
assert gateway.request("PUT", path, config, key=member_key).status_code == 403
assert gateway.request("DELETE", path, key=member_key).status_code == 403
finally:
assert gateway.request("DELETE", path, key=admin_key).status_code == 204

View file

@ -0,0 +1,67 @@
from datetime import date
from typing import Final
from urllib.parse import parse_qs, urlsplit
import pytest
from fastapi import HTTPException
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.alerting_endpoints.usage_alerts import require_scope
from litellm.repositories.usage_alerts import UsageAlertsRepository, activity_path
from litellm.types.proxy.auth.user_roles import LitellmUserRoles
class NoDatabase:
async def query_raw(self, query: str, *args: object) -> object:
raise AssertionError("Unauthorized callers must not reach the database")
@pytest.mark.asyncio
@pytest.mark.parametrize(
"role,key_type,write",
[
(LitellmUserRoles.PROXY_ADMIN, "llm_api", False),
(LitellmUserRoles.PROXY_ADMIN, "read_only", True),
(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY, None, True),
(LitellmUserRoles.INTERNAL_USER_VIEW_ONLY, None, True),
(LitellmUserRoles.INTERNAL_USER, None, False),
(LitellmUserRoles.CUSTOMER, None, False),
],
)
async def test_global_access_rejects_inference_keys_and_unprivileged_roles(
role: LitellmUserRoles, key_type: str | None, write: bool
) -> None:
user: Final = UserAPIKeyAuth(user_id="qa-user", user_role=role, key_type=key_type)
with pytest.raises(HTTPException) as error:
await require_scope(user, UsageAlertsRepository(NoDatabase()), "deployment", None, write=write)
assert error.value.status_code == 403
@pytest.mark.asyncio
async def test_internal_user_can_read_and_manage_only_their_user_scope() -> None:
user: Final = UserAPIKeyAuth(user_id="qa-user", user_role=LitellmUserRoles.INTERNAL_USER)
repository: Final = UsageAlertsRepository(NoDatabase())
await require_scope(user, repository, "user", "qa-user", write=True)
with pytest.raises(HTTPException) as error:
await require_scope(user, repository, "user", "other-user", write=False)
assert error.value.status_code == 403
def test_activity_link_escapes_entity_identity_and_includes_baseline() -> None:
path: Final = activity_path("model", "provider/model?foo=bar&other=1", date(2026, 11, 4), "qa-alert", 7)
parsed: Final = urlsplit(path)
query: Final = parse_qs(parsed.query)
assert parsed.path == "/models-and-endpoints"
assert query["activity_model"] == ["provider/model?foo=bar&other=1"]
assert query["activity_from"] == ["2026-10-28"]
assert query["activity_to"] == ["2026-11-04"]
assert query["alert"] == ["qa-alert"]
assert query["detail_tab"] == ["activity"]
def test_global_activity_link_uses_the_usage_page_date_contract() -> None:
query: Final = parse_qs(urlsplit(activity_path("deployment", None, date(2026, 11, 4), "qa-alert")).query)
assert query["tab"] == ["keys"]
assert "range" not in query
assert query["from"] == ["2026-10-28"]
assert query["to"] == ["2026-11-04"]

View file

@ -0,0 +1,43 @@
from typing import Final
import pytest
from pydantic import ValidationError
from litellm.types.proxy.usage_alerts import UsageAlertRuleInput
RULE: Final = {
"name": "Daily spend",
"scope_type": "team",
"scope_id": "qa-team",
"metric": "spend",
"direction": "increase",
}
@pytest.mark.parametrize(
"changes",
[
{"multiplier": 1},
{"multiplier": float("inf")},
{"min_value": float("nan")},
{"baseline_days": 2, "min_history_days": 3},
{"baseline_days": 31},
{"scope_type": "deployment"},
{"scope_type": "key", "scope_id": "sk-test-placeholder"},
{"notification_channels": ["email"]},
{"notification_channels": ["slack", "slack"]},
{"email_recipients": ["ops@example.com\r\nBcc: other@example.com"]},
],
)
def test_invalid_rules_cannot_be_stored(changes: dict[str, object]) -> None:
with pytest.raises(ValidationError):
UsageAlertRuleInput.model_validate({**RULE, **changes})
def test_zero_request_floor_and_daily_drop_multiplier_are_supported() -> None:
rule: Final = UsageAlertRuleInput.model_validate(
{**RULE, "metric": "requests", "direction": "decrease", "multiplier": 2, "min_requests": 0}
)
assert rule.direction == "decrease"
assert rule.multiplier == 2
assert rule.min_requests == 0

View file

@ -18834,6 +18834,78 @@ export interface paths {
patch?: never;
trace?: never;
};
"/usage/alert-rules": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
/** List Usage Alert Rules */
get: operations["list_usage_alert_rules_usage_alert_rules_get"];
put?: never;
/** Create Usage Alert Rule */
post: operations["create_usage_alert_rule_usage_alert_rules_post"];
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/usage/alert-rules/{rule_id}": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
/** Get Usage Alert Rule */
get: operations["get_usage_alert_rule_usage_alert_rules__rule_id__get"];
/** Update Usage Alert Rule */
put: operations["update_usage_alert_rule_usage_alert_rules__rule_id__put"];
post?: never;
/** Delete Usage Alert Rule */
delete: operations["delete_usage_alert_rule_usage_alert_rules__rule_id__delete"];
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/usage/alerts": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
/** List Usage Alerts */
get: operations["list_usage_alerts_usage_alerts_get"];
put?: never;
post?: never;
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/usage/alerts/{alert_id}": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
/** Get Usage Alert */
get: operations["get_usage_alert_usage_alerts__alert_id__get"];
put?: never;
post?: never;
delete?: never;
options?: never;
head?: never;
/** Update Usage Alert */
patch: operations["update_usage_alert_usage_alerts__alert_id__patch"];
trace?: never;
};
"/user/available_roles": {
parameters: {
query?: never;
@ -50331,6 +50403,250 @@ export interface components {
*/
version: string | null;
};
/** UsageAlert */
UsageAlert: {
/** Activity Path */
activity_path: string;
/**
* Baseline End
* Format: date
*/
baseline_end: string;
/** Baseline Sample Days */
baseline_sample_days?: number | null;
/**
* Baseline Start
* Format: date
*/
baseline_start: string;
/** Baseline Value */
baseline_value: number;
/**
* Day
* Format: date
*/
day: string;
/**
* Direction
* @enum {string}
*/
direction: "increase" | "decrease";
/**
* First Seen
* Format: date-time
*/
first_seen: string;
/** Id */
id: string;
/**
* Last Seen
* Format: date-time
*/
last_seen: string;
/**
* Metric
* @enum {string}
*/
metric: "spend" | "requests" | "latency" | "errors";
/** Observed Value */
observed_value: number;
/** Rule Id */
rule_id: string;
/** Rule Name */
rule_name: string;
/** Scope Id */
scope_id: string | null;
/**
* Scope Type
* @enum {string}
*/
scope_type: "deployment" | "key" | "user" | "team" | "model";
/**
* Status
* @enum {string}
*/
status: "open" | "resolved" | "dismissed";
/** Threshold */
threshold: number;
};
/** UsageAlertList */
UsageAlertList: {
/** Data */
data: components["schemas"]["UsageAlert"][];
/** Limit */
limit: number;
/** Offset */
offset: number;
/** Total */
total: number;
};
/** UsageAlertRule */
UsageAlertRule: {
/**
* Baseline Days
* @default 7
*/
baseline_days: number;
/**
* Created At
* Format: date-time
*/
created_at: string;
/** Created By */
created_by: string | null;
/**
* Direction
* @enum {string}
*/
direction: "increase" | "decrease";
/**
* Email Recipients
* @default []
*/
email_recipients: string[];
/**
* Enabled
* @default true
*/
enabled: boolean;
/**
* Evaluation Status
* @default pending
* @enum {string}
*/
evaluation_status: "pending" | "complete" | "insufficient_coverage" | "insufficient_history";
/** Id */
id: string;
/** Last Evaluated Day */
last_evaluated_day?: string | null;
/**
* Metric
* @enum {string}
*/
metric: "spend" | "requests" | "latency" | "errors";
/**
* Min History Days
* @default 3
*/
min_history_days: number;
/**
* Min Requests
* @default 100
*/
min_requests: number;
/**
* Min Value
* @description Minimum absolute change in the metric's units
* @default 0
*/
min_value: number;
/**
* Multiplier
* @default 3
*/
multiplier: number;
/** Name */
name: string;
/**
* Notification Channels
* @default []
*/
notification_channels: ("email" | "slack")[];
/** Scope Id */
scope_id?: string | null;
/**
* Scope Type
* @enum {string}
*/
scope_type: "deployment" | "key" | "user" | "team" | "model";
/**
* Updated At
* Format: date-time
*/
updated_at: string;
};
/** UsageAlertRuleInput */
UsageAlertRuleInput: {
/**
* Baseline Days
* @default 7
*/
baseline_days: number;
/**
* Direction
* @enum {string}
*/
direction: "increase" | "decrease";
/**
* Email Recipients
* @default []
*/
email_recipients: string[];
/**
* Enabled
* @default true
*/
enabled: boolean;
/**
* Metric
* @enum {string}
*/
metric: "spend" | "requests" | "latency" | "errors";
/**
* Min History Days
* @default 3
*/
min_history_days: number;
/**
* Min Requests
* @default 100
*/
min_requests: number;
/**
* Min Value
* @description Minimum absolute change in the metric's units
* @default 0
*/
min_value: number;
/**
* Multiplier
* @default 3
*/
multiplier: number;
/** Name */
name: string;
/**
* Notification Channels
* @default []
*/
notification_channels: ("email" | "slack")[];
/** Scope Id */
scope_id?: string | null;
/**
* Scope Type
* @enum {string}
*/
scope_type: "deployment" | "key" | "user" | "team" | "model";
};
/** UsageAlertRuleList */
UsageAlertRuleList: {
/** Data */
data: components["schemas"]["UsageAlertRule"][];
/** Limit */
limit: number;
/** Offset */
offset: number;
/** Total */
total: number;
};
/** UsageAlertUpdate */
UsageAlertUpdate: {
/**
* Status
* @enum {string}
*/
status: "open" | "resolved" | "dismissed";
};
/** UsageChartPoint */
UsageChartPoint: {
/** Blocked */
@ -78242,6 +78558,272 @@ export interface operations {
};
};
};
list_usage_alert_rules_usage_alert_rules_get: {
parameters: {
query?: {
scope_type?: ("deployment" | "key" | "user" | "team" | "model") | null;
scope_id?: string | null;
limit?: number;
offset?: number;
};
header?: never;
path?: never;
cookie?: never;
};
requestBody?: never;
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["UsageAlertRuleList"];
};
};
/** @description Validation Error */
422: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["HTTPValidationError"];
};
};
};
};
create_usage_alert_rule_usage_alert_rules_post: {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
requestBody: {
content: {
"application/json": components["schemas"]["UsageAlertRuleInput"];
};
};
responses: {
/** @description Successful Response */
201: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["UsageAlertRule"];
};
};
/** @description Validation Error */
422: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["HTTPValidationError"];
};
};
};
};
get_usage_alert_rule_usage_alert_rules__rule_id__get: {
parameters: {
query?: never;
header?: never;
path: {
rule_id: string;
};
cookie?: never;
};
requestBody?: never;
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["UsageAlertRule"];
};
};
/** @description Validation Error */
422: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["HTTPValidationError"];
};
};
};
};
update_usage_alert_rule_usage_alert_rules__rule_id__put: {
parameters: {
query?: never;
header?: never;
path: {
rule_id: string;
};
cookie?: never;
};
requestBody: {
content: {
"application/json": components["schemas"]["UsageAlertRuleInput"];
};
};
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["UsageAlertRule"];
};
};
/** @description Validation Error */
422: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["HTTPValidationError"];
};
};
};
};
delete_usage_alert_rule_usage_alert_rules__rule_id__delete: {
parameters: {
query?: never;
header?: never;
path: {
rule_id: string;
};
cookie?: never;
};
requestBody?: never;
responses: {
/** @description Successful Response */
204: {
headers: {
[name: string]: unknown;
};
content?: never;
};
/** @description Validation Error */
422: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["HTTPValidationError"];
};
};
};
};
list_usage_alerts_usage_alerts_get: {
parameters: {
query?: {
scope_type?: ("deployment" | "key" | "user" | "team" | "model") | null;
scope_id?: string | null;
metric?: ("spend" | "requests" | "latency" | "errors") | null;
status?: ("open" | "resolved" | "dismissed") | null;
start_date?: string | null;
end_date?: string | null;
limit?: number;
offset?: number;
};
header?: never;
path?: never;
cookie?: never;
};
requestBody?: never;
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["UsageAlertList"];
};
};
/** @description Validation Error */
422: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["HTTPValidationError"];
};
};
};
};
get_usage_alert_usage_alerts__alert_id__get: {
parameters: {
query?: never;
header?: never;
path: {
alert_id: string;
};
cookie?: never;
};
requestBody?: never;
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["UsageAlert"];
};
};
/** @description Validation Error */
422: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["HTTPValidationError"];
};
};
};
};
update_usage_alert_usage_alerts__alert_id__patch: {
parameters: {
query?: never;
header?: never;
path: {
alert_id: string;
};
cookie?: never;
};
requestBody: {
content: {
"application/json": components["schemas"]["UsageAlertUpdate"];
};
};
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["UsageAlert"];
};
};
/** @description Validation Error */
422: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["HTTPValidationError"];
};
};
};
};
ui_get_available_role_user_available_roles_get: {
parameters: {
query?: never;