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

feat(usage): aggregated user daily activity endpoint and UI integration (#13395) (#13400)

Browse source 
* feat(usage): add aggregated user daily activity endpoint and UI integration; fallback to paginated flow if unavailable

* refactor(usage): deduplicate daily activity logic; add wrapper to user paginated endpoint; share date formatting in UI

* chore(lint): remove unused imports from internal_user_endpoints

Co-authored-by: Cole McIntosh <82463175+colesmcintosh@users.noreply.github.com>
This commit is contained in:
Ishaan Jaff 2025-08-07 21:10:40 -07:00 • committed by GitHub
parent 43af255b45
commit a5cf880acc
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 350 additions and 197 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

306
litellm/proxy/management_endpoints/common_daily_activity.py
View file

@ -267,6 +267,106 @@ async def get_api_key_metadata(
}
def _build_where_conditions(
*,
entity_id_field: str,
entity_id: Optional[Union[str, List[str]]],
start_date: str,
end_date: str,
model: Optional[str],
api_key: Optional[str],
exclude_entity_ids: Optional[List[str]] = None,
) -> Dict[str, Any]:
"""Build prisma where clause for daily activity queries."""
where_conditions: Dict[str, Any] = {
"date": {
"gte": start_date,
"lte": end_date,
}
}
if model:
where_conditions["model"] = model
if api_key:
where_conditions["api_key"] = api_key
if entity_id is not None:
if isinstance(entity_id, list):
where_conditions[entity_id_field] = {"in": entity_id}
else:
where_conditions[entity_id_field] = {"equals": entity_id}
if exclude_entity_ids:
current = where_conditions.get(entity_id_field, {})
if isinstance(current, str):
current = {"equals": current}
current["not"] = {"in": exclude_entity_ids}
where_conditions[entity_id_field] = current
return where_conditions
async def _aggregate_spend_records(
*,
prisma_client: PrismaClient,
records: List[Any],
entity_id_field: Optional[str],
entity_metadata_field: Optional[Dict[str, dict]],
) -> Dict[str, Any]:
"""Aggregate rows into DailySpendData list and total metrics."""
api_keys: Set[str] = set()
for record in records:
if record.api_key:
api_keys.add(record.api_key)
api_key_metadata: Dict[str, Dict[str, Any]] = {}
model_metadata: Dict[str, Dict[str, Any]] = {}
provider_metadata: Dict[str, Dict[str, Any]] = {}
if api_keys:
api_key_metadata = await get_api_key_metadata(prisma_client, api_keys)
results: List[DailySpendData] = []
total_metrics = SpendMetrics()
grouped_data: Dict[str, Dict[str, Any]] = {}
for record in records:
date_str = record.date
if date_str not in grouped_data:
grouped_data[date_str] = {
"metrics": SpendMetrics(),
"breakdown": BreakdownMetrics(),
}
grouped_data[date_str]["metrics"] = update_metrics(
grouped_data[date_str]["metrics"], record
)
grouped_data[date_str]["breakdown"] = update_breakdown_metrics(
grouped_data[date_str]["breakdown"],
record,
model_metadata,
provider_metadata,
api_key_metadata,
entity_id_field=entity_id_field,
entity_metadata_field=entity_metadata_field,
)
total_metrics = update_metrics(total_metrics, record)
for date_str, data in grouped_data.items():
results.append(
DailySpendData(
date=datetime.strptime(date_str, "%Y-%m-%d").date(),
metrics=data["metrics"],
breakdown=data["breakdown"],
)
)
results.sort(key=lambda x: x.date, reverse=True)
return {"results": results, "totals": total_metrics}
async def get_daily_activity(
prisma_client: Optional[PrismaClient],
table_name: str,
@ -296,27 +396,15 @@ async def get_daily_activity(
)
try:
# Build filter conditions
where_conditions: Dict[str, Any] = {
"date": {
"gte": start_date,
"lte": end_date,
}
}
if model:
where_conditions["model"] = model
if api_key:
where_conditions["api_key"] = api_key
if entity_id is not None:
if isinstance(entity_id, list):
where_conditions[entity_id_field] = {"in": entity_id}
else:
where_conditions[entity_id_field] = entity_id
if exclude_entity_ids:
where_conditions.setdefault(entity_id_field, {})["not"] = {
"in": exclude_entity_ids
}
where_conditions = _build_where_conditions(
entity_id_field=entity_id_field,
entity_id=entity_id,
start_date=start_date,
end_date=end_date,
model=model,
api_key=api_key,
exclude_entity_ids=exclude_entity_ids,
)
# Get total count for pagination
total_count = await getattr(prisma_client.db, table_name).count(
@ -333,87 +421,25 @@ async def get_daily_activity(
take=page_size,
)
# Get all unique API keys from the spend data
api_keys = set()
for record in daily_spend_data:
if record.api_key:
api_keys.add(record.api_key)
# Fetch key aliases in bulk
api_key_metadata: Dict[str, Dict[str, Any]] = {}
model_metadata: Dict[str, Dict[str, Any]] = {}
provider_metadata: Dict[str, Dict[str, Any]] = {}
if api_keys:
api_key_metadata = await get_api_key_metadata(prisma_client, api_keys)
# Process results
results = []
total_metrics = SpendMetrics()
grouped_data: Dict[str, Dict[str, Any]] = {}
for record in daily_spend_data:
date_str = record.date
if date_str not in grouped_data:
grouped_data[date_str] = {
"metrics": SpendMetrics(),
"breakdown": BreakdownMetrics(),
}
# Update metrics
grouped_data[date_str]["metrics"] = update_metrics(
grouped_data[date_str]["metrics"], record
)
# Update breakdowns
grouped_data[date_str]["breakdown"] = update_breakdown_metrics(
grouped_data[date_str]["breakdown"],
record,
model_metadata,
provider_metadata,
api_key_metadata,
entity_id_field=entity_id_field,
entity_metadata_field=entity_metadata_field,
)
# Update total metrics
total_metrics.spend += record.spend
total_metrics.prompt_tokens += record.prompt_tokens
total_metrics.completion_tokens += record.completion_tokens
total_metrics.total_tokens += (
record.prompt_tokens + record.completion_tokens
)
total_metrics.cache_read_input_tokens += record.cache_read_input_tokens
total_metrics.cache_creation_input_tokens += (
record.cache_creation_input_tokens
)
total_metrics.api_requests += record.api_requests
total_metrics.successful_requests += record.successful_requests
total_metrics.failed_requests += record.failed_requests
# Convert grouped data to response format
for date_str, data in grouped_data.items():
results.append(
DailySpendData(
date=datetime.strptime(date_str, "%Y-%m-%d").date(),
metrics=data["metrics"],
breakdown=data["breakdown"],
)
)
# Sort results by date
results.sort(key=lambda x: x.date, reverse=True)
aggregated = await _aggregate_spend_records(
prisma_client=prisma_client,
records=daily_spend_data,
entity_id_field=entity_id_field,
entity_metadata_field=entity_metadata_field,
)
return SpendAnalyticsPaginatedResponse(
results=results,
results=aggregated["results"],
metadata=DailySpendMetadata(
total_spend=total_metrics.spend,
total_prompt_tokens=total_metrics.prompt_tokens,
total_completion_tokens=total_metrics.completion_tokens,
total_tokens=total_metrics.total_tokens,
total_api_requests=total_metrics.api_requests,
total_successful_requests=total_metrics.successful_requests,
total_failed_requests=total_metrics.failed_requests,
total_cache_read_input_tokens=total_metrics.cache_read_input_tokens,
total_cache_creation_input_tokens=total_metrics.cache_creation_input_tokens,
total_spend=aggregated["totals"].spend,
total_prompt_tokens=aggregated["totals"].prompt_tokens,
total_completion_tokens=aggregated["totals"].completion_tokens,
total_tokens=aggregated["totals"].total_tokens,
total_api_requests=aggregated["totals"].api_requests,
total_successful_requests=aggregated["totals"].successful_requests,
total_failed_requests=aggregated["totals"].failed_requests,
total_cache_read_input_tokens=aggregated["totals"].cache_read_input_tokens,
total_cache_creation_input_tokens=aggregated["totals"].cache_creation_input_tokens,
page=page,
total_pages=-(-total_count // page_size), # Ceiling division
has_more=(page * page_size) < total_count,
@ -426,3 +452,85 @@ async def get_daily_activity(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail={"error": f"Failed to fetch analytics: {str(e)}"},
)
async def get_daily_activity_aggregated(
prisma_client: Optional[PrismaClient],
table_name: str,
entity_id_field: str,
entity_id: Optional[Union[str, List[str]]],
entity_metadata_field: Optional[Dict[str, dict]],
start_date: Optional[str],
end_date: Optional[str],
model: Optional[str],
api_key: Optional[str],
exclude_entity_ids: Optional[List[str]] = None,
) -> SpendAnalyticsPaginatedResponse:
"""Aggregated variant that returns the full result set (no pagination).
Matches the response model of the paginated endpoint so the UI does not need to transform.
"""
if prisma_client is None:
raise HTTPException(
status_code=500,
detail={"error": CommonProxyErrors.db_not_connected_error.value},
)
if start_date is None or end_date is None:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={"error": "Please provide start_date and end_date"},
)
try:
where_conditions = _build_where_conditions(
entity_id_field=entity_id_field,
entity_id=entity_id,
start_date=start_date,
end_date=end_date,
model=model,
api_key=api_key,
exclude_entity_ids=exclude_entity_ids,
)
# Fetch all matching results (no pagination)
daily_spend_data = await getattr(prisma_client.db, table_name).find_many(
where=where_conditions,
order=[
{"date": "desc"},
],
)
aggregated = await _aggregate_spend_records(
prisma_client=prisma_client,
records=daily_spend_data,
entity_id_field=entity_id_field,
entity_metadata_field=entity_metadata_field,
)
return SpendAnalyticsPaginatedResponse(
results=aggregated["results"],
metadata=DailySpendMetadata(
total_spend=aggregated["totals"].spend,
total_prompt_tokens=aggregated["totals"].prompt_tokens,
total_completion_tokens=aggregated["totals"].completion_tokens,
total_tokens=aggregated["totals"].total_tokens,
total_api_requests=aggregated["totals"].api_requests,
total_successful_requests=aggregated["totals"].successful_requests,
total_failed_requests=aggregated["totals"].failed_requests,
total_cache_read_input_tokens=aggregated["totals"].cache_read_input_tokens,
total_cache_creation_input_tokens=aggregated["totals"].cache_creation_input_tokens,
page=1,
total_pages=1,
has_more=False,
),
)
except Exception as e:
verbose_proxy_logger.exception(
f"Error fetching aggregated daily activity: {str(e)}"
)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail={"error": f"Failed to fetch analytics: {str(e)}"},
)

149
litellm/proxy/management_endpoints/internal_user_endpoints.py
View file

@ -26,7 +26,10 @@ from litellm._logging import verbose_proxy_logger
from litellm.proxy._types import *
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.hooks.user_management_event_hooks import UserManagementEventHooks
from litellm.proxy.management_endpoints.common_daily_activity import get_daily_activity
from litellm.proxy.management_endpoints.common_daily_activity import (
get_daily_activity,
get_daily_activity_aggregated,
)
from litellm.proxy.management_endpoints.common_utils import _user_has_admin_view
from litellm.proxy.management_endpoints.key_management_endpoints import (
generate_key_helper_fn,
@ -35,13 +38,7 @@ from litellm.proxy.management_endpoints.key_management_endpoints import (
from litellm.proxy.management_helpers.utils import management_endpoint_wrapper
from litellm.proxy.utils import handle_exception_on_proxy
from litellm.types.proxy.management_endpoints.common_daily_activity import (
BreakdownMetrics,
KeyMetadata,
KeyMetricWithMetadata,
LiteLLM_DailyUserSpend,
MetricWithMetadata,
SpendAnalyticsPaginatedResponse,
SpendMetrics,
)
from litellm.types.proxy.management_endpoints.internal_user_endpoints import (
BulkUpdateUserRequest,
@ -1784,71 +1781,7 @@ async def ui_view_users(
raise HTTPException(status_code=500, detail=f"Error searching users: {str(e)}")
def update_metrics(
group_metrics: SpendMetrics, record: LiteLLM_DailyUserSpend
) -> SpendMetrics:
group_metrics.spend += record.spend
group_metrics.prompt_tokens += record.prompt_tokens
group_metrics.completion_tokens += record.completion_tokens
group_metrics.cache_read_input_tokens += record.cache_read_input_tokens
group_metrics.cache_creation_input_tokens += record.cache_creation_input_tokens
group_metrics.total_tokens += record.prompt_tokens + record.completion_tokens
group_metrics.api_requests += record.api_requests
group_metrics.successful_requests += record.successful_requests
group_metrics.failed_requests += record.failed_requests
return group_metrics
def update_breakdown_metrics(
breakdown: BreakdownMetrics,
record: LiteLLM_DailyUserSpend,
model_metadata: Dict[str, Dict[str, Any]],
provider_metadata: Dict[str, Dict[str, Any]],
api_key_metadata: Dict[str, Dict[str, Any]],
) -> BreakdownMetrics:
"""Updates breakdown metrics for a single record using the existing update_metrics function"""
# Update model breakdown
if record.model:
if record.model not in breakdown.models:
breakdown.models[record.model] = MetricWithMetadata(
metrics=SpendMetrics(),
metadata=model_metadata.get(
record.model, {}
), # Add any model-specific metadata here
)
breakdown.models[record.model].metrics = update_metrics(
breakdown.models[record.model].metrics, record
)
# Update provider breakdown
provider = record.custom_llm_provider or "unknown"
if provider not in breakdown.providers:
breakdown.providers[provider] = MetricWithMetadata(
metrics=SpendMetrics(),
metadata=provider_metadata.get(
provider, {}
), # Add any provider-specific metadata here
)
breakdown.providers[provider].metrics = update_metrics(
breakdown.providers[provider].metrics, record
)
# Update api key breakdown
if record.api_key not in breakdown.api_keys:
breakdown.api_keys[record.api_key] = KeyMetricWithMetadata(
metrics=SpendMetrics(),
metadata=KeyMetadata(
key_alias=api_key_metadata.get(record.api_key, {}).get(
"key_alias", None
)
), # Add any api_key-specific metadata here
)
breakdown.api_keys[record.api_key].metrics = update_metrics(
breakdown.api_keys[record.api_key].metrics, record
)
return breakdown
# Using shared metric helper implementations from common_daily_activity
@router.get(
@ -1857,6 +1790,7 @@ def update_breakdown_metrics(
dependencies=[Depends(user_api_key_auth)],
response_model=SpendAnalyticsPaginatedResponse,
)
@management_endpoint_wrapper
async def get_user_daily_activity(
start_date: Optional[str] = fastapi.Query(
default=None,
@ -1939,3 +1873,74 @@ async def get_user_daily_activity(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail={"error": f"Failed to fetch analytics: {str(e)}"},
)
@router.get(
"/user/daily/activity/aggregated",
tags=["Budget & Spend Tracking", "Internal User management"],
dependencies=[Depends(user_api_key_auth)],
response_model=SpendAnalyticsPaginatedResponse,
)
@management_endpoint_wrapper
async def get_user_daily_activity_aggregated(
start_date: Optional[str] = fastapi.Query(
default=None,
description="Start date in YYYY-MM-DD format",
),
end_date: Optional[str] = fastapi.Query(
default=None,
description="End date in YYYY-MM-DD format",
),
model: Optional[str] = fastapi.Query(
default=None,
description="Filter by specific model",
),
api_key: Optional[str] = fastapi.Query(
default=None,
description="Filter by specific API key",
),
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
) -> SpendAnalyticsPaginatedResponse:
"""
Aggregated analytics for a user's daily activity without pagination.
Returns the same response shape as the paginated endpoint with page metadata set to single-page.
"""
from litellm.proxy.proxy_server import prisma_client
if prisma_client is None:
raise HTTPException(
status_code=500,
detail={"error": CommonProxyErrors.db_not_connected_error.value},
)
if start_date is None or end_date is None:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={"error": "Please provide start_date and end_date"},
)
try:
entity_id: Optional[str] = None
if not _user_has_admin_view(user_api_key_dict):
entity_id = user_api_key_dict.user_id
return await get_daily_activity_aggregated(
prisma_client=prisma_client,
table_name="litellm_dailyuserspend",
entity_id_field="user_id",
entity_id=entity_id,
entity_metadata_field=None,
start_date=start_date,
end_date=end_date,
model=model,
api_key=api_key,
)
except Exception as e:
verbose_proxy_logger.exception(
"/user/daily/activity/aggregated: Exception occured - {}".format(str(e))
)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail={"error": f"Failed to fetch analytics: {str(e)}"},
)

77
ui/litellm-dashboard/src/components/networking.tsx
View file

@ -1,3 +1,10 @@
// Shared date formatter for daily activity endpoints
export const formatDate = (date: Date) => {
const year = date.getFullYear();
const month = String(date.getMonth() + 1).padStart(2, '0');
const day = String(date.getDate()).padStart(2, '0');
return `${year}-${month}-${day}`;
};
/**
* Helper file for calls being made to proxy
*/
@ -1457,13 +1464,6 @@ export const userDailyActivityCall = async (
? `${proxyBaseUrl}/user/daily/activity`
: `/user/daily/activity`;
const queryParams = new URLSearchParams();
// Format dates as YYYY-MM-DD for the API
const formatDate = (date: Date) => {
const year = date.getFullYear();
const month = String(date.getMonth() + 1).padStart(2, '0');
const day = String(date.getDate()).padStart(2, '0');
return `${year}-${month}-${day}`;
};
queryParams.append("start_date", formatDate(startTime));
queryParams.append("end_date", formatDate(endTime));
queryParams.append("page_size", "1000");
@ -1510,13 +1510,6 @@ export const tagDailyActivityCall = async (
? `${proxyBaseUrl}/tag/daily/activity`
: `/tag/daily/activity`;
const queryParams = new URLSearchParams();
// Format dates as YYYY-MM-DD for the API
const formatDate = (date: Date) => {
const year = date.getFullYear();
const month = String(date.getMonth() + 1).padStart(2, '0');
const day = String(date.getDate()).padStart(2, '0');
return `${year}-${month}-${day}`;
};
queryParams.append("start_date", formatDate(startTime));
queryParams.append("end_date", formatDate(endTime));
queryParams.append("page_size", "1000");
@ -1566,13 +1559,6 @@ export const teamDailyActivityCall = async (
? `${proxyBaseUrl}/team/daily/activity`
: `/team/daily/activity`;
const queryParams = new URLSearchParams();
// Format dates as YYYY-MM-DD for the API
const formatDate = (date: Date) => {
const year = date.getFullYear();
const month = String(date.getMonth() + 1).padStart(2, '0');
const day = String(date.getDate()).padStart(2, '0');
return `${year}-${month}-${day}`;
};
queryParams.append("start_date", formatDate(startTime));
queryParams.append("end_date", formatDate(endTime));
queryParams.append("page_size", "1000");
@ -3198,6 +3184,55 @@ export interface User {
[key: string]: string; // Include any other potential keys in the dictionary
}
export const userDailyActivityAggregatedCall = async (
accessToken: String,
startTime: Date,
endTime: Date
) => {
/**
* Get aggregated daily user activity (no pagination)
*/
try {
let url = proxyBaseUrl
? `${proxyBaseUrl}/user/daily/activity/aggregated`
: `/user/daily/activity/aggregated`;
const queryParams = new URLSearchParams();
// Format dates as YYYY-MM-DD for the API
const formatDate = (date: Date) => {
const year = date.getFullYear();
const month = String(date.getMonth() + 1).padStart(2, '0');
const day = String(date.getDate()).padStart(2, '0');
return `${year}-${month}-${day}`;
};
queryParams.append("start_date", formatDate(startTime));
queryParams.append("end_date", formatDate(endTime));
const queryString = queryParams.toString();
if (queryString) {
url += `?${queryString}`;
}
const response = await fetch(url, {
method: "GET",
headers: {
[globalLitellmHeaderName]: `Bearer ${accessToken}`,
"Content-Type": "application/json",
},
});
if (!response.ok) {
const errorData = await response.text();
handleError(errorData);
throw new Error("Network response was not ok");
}
const data = await response.json();
return data;
} catch (error) {
console.error("Failed to fetch aggregated user daily activity:", error);
throw error;
}
};
export const userGetAllUsersCall = async (
accessToken: String,
role: String

15
ui/litellm-dashboard/src/components/new_usage.tsx
View file

@ -34,7 +34,7 @@ import {
import AdvancedDatePicker from "./shared/advanced_date_picker"
import { AreaChart } from "@tremor/react"
import { userDailyActivityCall, tagListCall } from "./networking"
import { userDailyActivityCall, userDailyActivityAggregatedCall, tagListCall } from "./networking"
import { Tag } from "./tag_management/types"
import ViewUserSpend from "./view_user_spend"
import TopKeyView from "./top_key_view"
@ -304,16 +304,22 @@ const NewUsagePage: React.FC<NewUsagePageProps> = ({ accessToken, userRole, user
const endTime = new Date(dateValue.to)
try {
// Get first page
// Prefer aggregated endpoint to avoid many page requests
try {
const aggregated = await userDailyActivityAggregatedCall(accessToken, startTime, endTime)
setUserSpendData(aggregated)
return
} catch (e) {
// Fallback to paginated calls if aggregated endpoint is unavailable
}
const firstPageData = await userDailyActivityCall(accessToken, startTime, endTime)
// If only one page, just set the data
if (firstPageData.metadata.total_pages <= 1) {
setUserSpendData(firstPageData)
return
}
// Fetch all pages
const allResults = [...firstPageData.results]
const aggregatedMetadata = { ...firstPageData.metadata }
@ -329,7 +335,6 @@ const NewUsagePage: React.FC<NewUsagePageProps> = ({ accessToken, userRole, user
}
}
// Combine all results with the first page's metadata
setUserSpendData({
results: allResults,
metadata: aggregatedMetadata,