From a5cf880accba3561a1a79ffe2c3db8d820a73390 Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Thu, 7 Aug 2025 21:10:40 -0700 Subject: [PATCH] feat(usage): aggregated user daily activity endpoint and UI integration (#13395) (#13400) * 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> --- .../common_daily_activity.py | 306 ++++++++++++------ .../internal_user_endpoints.py | 149 ++++----- .../src/components/networking.tsx | 77 +++-- .../src/components/new_usage.tsx | 15 +- 4 files changed, 350 insertions(+), 197 deletions(-) diff --git a/litellm/proxy/management_endpoints/common_daily_activity.py b/litellm/proxy/management_endpoints/common_daily_activity.py index bdad6cd648a..dbf1cdf514c 100644 --- a/litellm/proxy/management_endpoints/common_daily_activity.py +++ b/litellm/proxy/management_endpoints/common_daily_activity.py @@ -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)}"}, + ) diff --git a/litellm/proxy/management_endpoints/internal_user_endpoints.py b/litellm/proxy/management_endpoints/internal_user_endpoints.py index e33ae24599c..3876230f5ab 100644 --- a/litellm/proxy/management_endpoints/internal_user_endpoints.py +++ b/litellm/proxy/management_endpoints/internal_user_endpoints.py @@ -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)}"}, + ) diff --git a/ui/litellm-dashboard/src/components/networking.tsx b/ui/litellm-dashboard/src/components/networking.tsx index 36a2ffe20aa..b8b84aa93d0 100644 --- a/ui/litellm-dashboard/src/components/networking.tsx +++ b/ui/litellm-dashboard/src/components/networking.tsx @@ -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 diff --git a/ui/litellm-dashboard/src/components/new_usage.tsx b/ui/litellm-dashboard/src/components/new_usage.tsx index ec9d462fb92..e17c2d78be3 100644 --- a/ui/litellm-dashboard/src/components/new_usage.tsx +++ b/ui/litellm-dashboard/src/components/new_usage.tsx @@ -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 = ({ 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 = ({ accessToken, userRole, user } } - // Combine all results with the first page's metadata setUserSpendData({ results: allResults, metadata: aggregatedMetadata,