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

feat(proxy): bounded daily activity routes for all usage entities (#43408)

Browse source 
Co-authored-by: yassin <yassin@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot] 2026-10-01 16:57:23 -07:00 • committed by GitHub
parent 339f5b6a4d
commit 54a51c80df
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
29 changed files with 11031 additions and 655 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
backend/routes/allowlist.py
View file

@ -60,6 +60,7 @@ BACKEND_PATH_PREFIXES: tuple[str, ...] = (
# Tools / agents (registry & policy admin)
"/v1/tool/",
"/v1/agents",
"/agent/daily/activity/",
# Guardrails admin
"/v2/guardrails/",
# MCP server admin + BYOK OAuth flow (UI-initiated) + dynamic per-server endpoints

1046
litellm/proxy/_lazy_openapi_snapshot.json
View file

File diff suppressed because it is too large Load diff

52
litellm/proxy/_types.py
View file

@ -699,6 +699,10 @@ class LiteLLMRoutes(enum.Enum):
KeyManagementRoutes.TEAM_KEY_BULK_UPDATE.value,
KeyManagementRoutes.TEAM_DAILY_ACTIVITY.value,
KeyManagementRoutes.TEAM_DAILY_ACTIVITY_AGGREGATED.value,
"/team/daily/activity/aggregated/keys",
"/team/daily/activity/aggregated/search",
"/team/daily/activity/aggregated/model_top_keys",
"/team/daily/activity/export",
KeyManagementRoutes.SPEND_LOGS.value,
KeyManagementRoutes.SPEND_LOGS_V2.value,
KeyManagementRoutes.KEY_RESET_SPEND.value,
@ -725,6 +729,11 @@ class LiteLLMRoutes(enum.Enum):
"/user/list",
"/user/daily/activity",
"/user/daily/activity/aggregated",
"/user/daily/activity/aggregated/keys",
"/user/daily/activity/aggregated/search",
"/user/daily/activity/aggregated/model_top_keys",
"/user/daily/activity/export",
"/user/daily/activity/aggregated/cache_leakage_keys",
# team
"/team/new",
"/team/update",
@ -742,6 +751,10 @@ class LiteLLMRoutes(enum.Enum):
"/team/permissions_bulk_update",
"/team/daily/activity",
"/team/daily/activity/aggregated",
"/team/daily/activity/aggregated/keys",
"/team/daily/activity/aggregated/search",
"/team/daily/activity/aggregated/model_top_keys",
"/team/daily/activity/export",
"/team/spend/by_user",
# gateway request counts (SGR); deployment-wide, admin-only
"/gateway/daily/activity",
@ -870,6 +883,11 @@ class LiteLLMRoutes(enum.Enum):
# Tag usage endpoints scope internal users to tags produced by
# their own keys in tag_management_endpoints.py.
"/tag/daily/activity",
"/tag/daily/activity/aggregated",
"/tag/daily/activity/aggregated/keys",
"/tag/daily/activity/aggregated/search",
"/tag/daily/activity/aggregated/model_top_keys",
"/tag/daily/activity/export",
"/tag/list",
"/v1/models/{model_id}",
"/models/{model_id}",
@ -894,6 +912,11 @@ class LiteLLMRoutes(enum.Enum):
# Tag usage endpoints scope internal viewers to tags produced by
# their own keys in tag_management_endpoints.py.
"/tag/daily/activity",
"/tag/daily/activity/aggregated",
"/tag/daily/activity/aggregated/keys",
"/tag/daily/activity/aggregated/search",
"/tag/daily/activity/aggregated/model_top_keys",
"/tag/daily/activity/export",
"/tag/list",
]
)
@ -913,6 +936,10 @@ class LiteLLMRoutes(enum.Enum):
"/team/permissions_update",
"/team/daily/activity",
"/team/daily/activity/aggregated",
"/team/daily/activity/aggregated/keys",
"/team/daily/activity/aggregated/search",
"/team/daily/activity/aggregated/model_top_keys",
"/team/daily/activity/export",
"/team/spend/by_user",
"/team/{team_id}/members/me",
# POST/GET the team's logging callbacks, and DELETE one of them. Every
@ -928,9 +955,19 @@ class LiteLLMRoutes(enum.Enum):
"/model/delete",
"/user/daily/activity",
"/user/daily/activity/aggregated",
"/user/daily/activity/aggregated/keys",
"/user/daily/activity/aggregated/search",
"/user/daily/activity/aggregated/model_top_keys",
"/user/daily/activity/export",
"/user/daily/activity/aggregated/cache_leakage_keys",
# Endpoint restricts results to organizations the caller is ORG_ADMIN
# of; a caller who administers none gets an empty result set.
"/organization/daily/activity",
"/organization/daily/activity/aggregated",
"/organization/daily/activity/aggregated/keys",
"/organization/daily/activity/aggregated/search",
"/organization/daily/activity/aggregated/model_top_keys",
"/organization/daily/activity/export",
"/user/available_roles", # read-only role metadata; any authenticated user may read
# Claude Code gateway: the signed-in CLI fetches its managed settings and posts its own telemetry
"/claude_code_gateway/managed/settings",
@ -1009,9 +1046,24 @@ class LiteLLMRoutes(enum.Enum):
"/user/available_users",
"/user/available_roles",
"/user/daily/activity",
"/user/daily/activity/aggregated",
"/user/daily/activity/aggregated/keys",
"/user/daily/activity/aggregated/search",
"/user/daily/activity/aggregated/model_top_keys",
"/user/daily/activity/export",
"/user/daily/activity/aggregated/cache_leakage_keys",
"/team/daily/activity",
"/team/daily/activity/aggregated",
"/team/daily/activity/aggregated/keys",
"/team/daily/activity/aggregated/search",
"/team/daily/activity/aggregated/model_top_keys",
"/team/daily/activity/export",
"/tag/daily/activity",
"/tag/daily/activity/aggregated",
"/tag/daily/activity/aggregated/keys",
"/tag/daily/activity/aggregated/search",
"/tag/daily/activity/aggregated/model_top_keys",
"/tag/daily/activity/export",
"/tag/list",
"/audit",
"/audit/{id}",

175
litellm/proxy/agent_endpoints/endpoints.py
View file

@ -13,7 +13,7 @@ import os
import uuid
from collections.abc import Mapping, Sequence
from types import MappingProxyType
from typing import Annotated, Final, TypedDict
from typing import Annotated, Final, NamedTuple, TypedDict
from fastapi import APIRouter, Depends, HTTPException, Query, Request
from pydantic import ValidationError
@ -47,7 +47,11 @@ from litellm.proxy.agent_endpoints.agent_search import (
global_agent_search_index,
search_agents,
)
from litellm.proxy.agent_endpoints.auth.agent_permission_handler import accessible_agents
from litellm.proxy.agent_endpoints.auth.agent_permission_handler import (
AgentRequestHandler,
UnrestrictedAgentAccess,
accessible_agents,
)
from litellm.proxy.agent_endpoints.identity import reject_legacy_identity
from litellm.proxy.agent_endpoints.identity_store import AgentIdentityStore
from litellm.proxy.agent_endpoints.kill_switch import (
@ -63,7 +67,8 @@ from litellm.proxy.agent_endpoints.managed_identity import raise_identity_failur
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.common_utils.rbac_utils import check_feature_access_for_user
from litellm.proxy.management_endpoints.common_daily_activity import get_daily_activity
from litellm.proxy.utils import get_custom_url
from litellm.proxy.utils import PrismaClient, get_custom_url
from litellm.repositories.chunked_in import find_many_in
from litellm.types.agents import (
AgentCard,
AgentConfig,
@ -303,6 +308,73 @@ async def _rank_agents_by_query(
assert_never(outcome)
class _AgentDailyActivityScope(NamedTuple):
agent_ids: tuple[str, ...] | None
agent_metadata: Mapping[str, dict[str, object]]
async def _owned_agent_ids(*, user_id: str | None, prisma_client: PrismaClient) -> frozenset[str]:
if user_id is None:
return frozenset()
owned_records: Final = await agents_table(prisma_client).find_many(where={"created_by": user_id})
return frozenset(agent.agent_id for agent in owned_records)
async def _permitted_daily_activity_agent_ids(
*, user_api_key_dict: UserAPIKeyAuth, prisma_client: PrismaClient
) -> frozenset[str]:
access: Final = await AgentRequestHandler.resolve_agent_access(user_api_key_auth=user_api_key_dict)
if isinstance(access, UnrestrictedAgentAccess):
return await _owned_agent_ids(user_id=user_api_key_dict.user_id, prisma_client=prisma_client)
return access.agent_ids
async def _resolve_daily_activity_agent_ids(
*,
agent_ids: tuple[str, ...] | None,
user_api_key_dict: UserAPIKeyAuth,
prisma_client: PrismaClient,
) -> tuple[str, ...] | None:
from litellm.proxy.management_endpoints.common_utils import _user_has_admin_view
if _user_has_admin_view(user_api_key_dict):
return agent_ids
permitted_agent_ids: Final = await _permitted_daily_activity_agent_ids(
user_api_key_dict=user_api_key_dict,
prisma_client=prisma_client,
)
return (
tuple(agent_id for agent_id in agent_ids if agent_id in permitted_agent_ids)
if agent_ids
else tuple(permitted_agent_ids)
)
async def resolve_agent_daily_activity_scope(
*,
agent_ids: tuple[str, ...] | None,
user_api_key_dict: UserAPIKeyAuth,
prisma_client: PrismaClient,
) -> _AgentDailyActivityScope:
await check_feature_access_for_user(user_api_key_dict, "agents")
resolved_agent_ids: Final = await _resolve_daily_activity_agent_ids(
agent_ids=agent_ids,
user_api_key_dict=user_api_key_dict,
prisma_client=prisma_client,
)
agent_records: Final = (
await agents_table(prisma_client).find_many(where={})
if resolved_agent_ids is None
else await find_many_in(agents_table(prisma_client), "agent_id", resolved_agent_ids)
)
agent_metadata: Final[Mapping[str, dict[str, object]]] = MappingProxyType(
{agent.agent_id: {"agent_name": agent.agent_name} for agent in agent_records}
)
return _AgentDailyActivityScope(resolved_agent_ids, agent_metadata)
@router.get(
"/v1/agents",
tags=["[beta] A2A Agents"],
@ -1288,82 +1360,39 @@ async def get_agent_daily_activity(
detail={"error": CommonProxyErrors.db_not_connected_error.value},
)
agent_ids_list = agent_ids.split(",") if agent_ids else None
exclude_agent_ids_list: list[str] | None = None
if exclude_agent_ids:
exclude_agent_ids_list = exclude_agent_ids.split(",") if exclude_agent_ids else None
# Without scoping, an empty `agent_ids` query returned every agent's
# spend/token rows on the proxy. Restrict non-admin callers to the
# agents they're permitted to invoke (or that they created), and
# intersect their explicit `agent_ids` filter with the same allowlist.
from litellm.proxy.agent_endpoints.auth.agent_permission_handler import (
AgentRequestHandler,
RestrictedAgentAccess,
UnrestrictedAgentAccess,
requested_agent_ids: Final = tuple(agent_ids.split(",")) if agent_ids else None
exclude_agent_ids_list: Final[list[str] | None] = exclude_agent_ids.split(",") if exclude_agent_ids else None
agent_scope: Final = await resolve_agent_daily_activity_scope(
agent_ids=requested_agent_ids,
user_api_key_dict=user_api_key_dict,
prisma_client=prisma_client,
)
from litellm.proxy.management_endpoints.common_utils import _user_has_admin_view
where_condition: Final[dict[str, object]] = {}
if not _user_has_admin_view(user_api_key_dict):
permitted_agent_ids: list[str] = []
# An unrestricted caller is not "see everything" for activity scoping. Fall
# back to the agents the caller created so they cannot enumerate other
# tenants' agents.
# Guard against `user_id is None`: a literal None in Prisma
# `where={"created_by": None}` resolves to ``created_by IS NULL``
# and would expose every ownerless agent's rows.
match await AgentRequestHandler.resolve_agent_access(user_api_key_auth=user_api_key_dict):
case RestrictedAgentAccess(allowed_agent_ids):
permitted_agent_ids = list(allowed_agent_ids)
case UnrestrictedAgentAccess():
if user_api_key_dict.user_id is not None:
owned_records: Final = await agents_table(prisma_client).find_many(
where={"created_by": user_api_key_dict.user_id}
)
permitted_agent_ids = [a.agent_id for a in owned_records]
if agent_ids_list:
permitted_agent_id_set: Final = set(permitted_agent_ids)
agent_ids_list = [aid for aid in agent_ids_list if aid in permitted_agent_id_set]
else:
agent_ids_list = list(permitted_agent_ids)
# No accessible agents → return an empty page without querying.
if not agent_ids_list:
return SpendAnalyticsPaginatedResponse(
results=[],
metadata=DailySpendMetadata(
total_spend=0.0,
total_prompt_tokens=0,
total_completion_tokens=0,
total_tokens=0,
total_api_requests=0,
total_successful_requests=0,
total_failed_requests=0,
total_cache_read_input_tokens=0,
total_cache_creation_input_tokens=0,
total_compression_saved_tokens=0,
page=page,
total_pages=0,
has_more=False,
),
)
if agent_ids_list:
where_condition["agent_id"] = {"in": list(agent_ids_list)}
agent_records: Final = await agents_table(prisma_client).find_many(where=where_condition)
agent_metadata: Final[Mapping[str, dict[str, object]]] = {
agent.agent_id: {"agent_name": agent.agent_name} for agent in agent_records
}
if agent_scope.agent_ids == ():
return SpendAnalyticsPaginatedResponse(
results=[],
metadata=DailySpendMetadata(
total_spend=0.0,
total_prompt_tokens=0,
total_completion_tokens=0,
total_tokens=0,
total_api_requests=0,
total_successful_requests=0,
total_failed_requests=0,
total_cache_read_input_tokens=0,
total_cache_creation_input_tokens=0,
total_compression_saved_tokens=0,
page=page,
total_pages=0,
has_more=False,
),
)
return await get_daily_activity(
prisma_client=prisma_client,
table_name="litellm_dailyagentspend",
entity_id_field="agent_id",
entity_id=agent_ids_list,
entity_metadata_field=agent_metadata,
entity_id=None if agent_scope.agent_ids is None else list(agent_scope.agent_ids),
entity_metadata_field=agent_scope.agent_metadata,
exclude_entity_ids=exclude_agent_ids_list,
start_date=start_date,
end_date=end_date,

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

@ -4,10 +4,10 @@ from collections.abc import Set as AbstractSet
from dataclasses import dataclass, replace
from datetime import datetime, timedelta
from types import MappingProxyType
from typing import Final, Protocol
from typing import Final, Literal, NoReturn, Protocol
from fastapi import HTTPException, status
from typing_extensions import ReadOnly, TypedDict
from typing_extensions import ReadOnly, TypedDict, assert_never
from litellm import constants
from litellm._logging import verbose_proxy_logger
@ -45,6 +45,27 @@ from litellm.types.repositories.daily_activity import (
)
@dataclass(frozen=True, slots=True)
class ScopeDenied:
status_code: Literal[403, 404]
reason: str
@dataclass(frozen=True, slots=True)
class InvalidDateRange:
reason: str
def raise_public(error: ScopeDenied | InvalidDateRange) -> NoReturn:
match error:
case ScopeDenied():
raise HTTPException(status_code=error.status_code, detail={"error": error.reason})
case InvalidDateRange():
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail={"error": error.reason})
case _:
assert_never(error)
class DailySpendRecord(Protocol):
@property
def date(self) -> str: ...
@ -402,7 +423,7 @@ def update_breakdown_metrics(
return breakdown
def _spend_logs_window(dates: AbstractSet[str | None]) -> tuple[datetime, datetime] | None:
def spend_logs_window(dates: AbstractSet[str | None]) -> tuple[datetime, datetime] | None:
parsed: Final = sorted(day for day in (_parse_spend_date(raw) for raw in dates) if day is not None)
if not parsed:
return None
@ -622,7 +643,7 @@ async def _aggregate_spend_records(
api_key_metadata: Final[Mapping[str, KeyMetadataRow]] = (
await repository.key_metadata(
frozenset(api_keys), _spend_logs_window(frozenset(record.date for record in records))
frozenset(api_keys), spend_logs_window(frozenset(record.date for record in records))
)
if api_keys
else MappingProxyType({})
@ -823,7 +844,7 @@ async def _aggregate_grouping_sets_records(
api_keys: Final[set[str]] = {r.api_key for r in records if r.api_key and r.api_key != PTU_SENTINEL_API_KEY}
api_key_metadata: Final[Mapping[str, KeyMetadataRow]] = (
await repository.key_metadata(frozenset(api_keys), _spend_logs_window(frozenset(r.date for r in records)))
await repository.key_metadata(frozenset(api_keys), spend_logs_window(frozenset(r.date for r in records)))
if api_keys
else MappingProxyType({})
)
@ -990,7 +1011,7 @@ async def get_daily_activity_aggregated(
)
entity_key_metadata: Final[Mapping[str, KeyMetadataRow]] = (
await repository.key_metadata(
entity_api_keys, _spend_logs_window(frozenset(row.date for row in entity_records))
entity_api_keys, spend_logs_window(frozenset(row.date for row in entity_records))
)
if entity_api_keys
else MappingProxyType({})

52
litellm/proxy/management_endpoints/customer_endpoints.py
View file

@ -12,7 +12,8 @@ All /customer management endpoints
#### END-USER/CUSTOMER MANAGEMENT ####
from collections.abc import Mapping, Sequence
from datetime import datetime, timedelta
from typing import TYPE_CHECKING, Final, Protocol, TypeVar, overload
from types import MappingProxyType
from typing import TYPE_CHECKING, Final, NamedTuple, Protocol, TypeVar, overload
import fastapi
from fastapi import APIRouter, Depends, HTTPException, Request
@ -41,6 +42,7 @@ from litellm.proxy.management_helpers.object_permission_utils import (
)
from litellm.proxy.utils import handle_exception_on_proxy
from litellm.repositories.budget_repository import BudgetRepository
from litellm.repositories.chunked_in import find_many_in
from litellm.repositories.table_repositories import EndUserRepository
from litellm.types.proxy.management_endpoints.common_daily_activity import (
SpendAnalyticsPaginatedResponse,
@ -490,6 +492,33 @@ async def new_end_user(
raise handle_exception_on_proxy(e)
class _CustomerDailyActivityScope(NamedTuple):
end_user_ids: tuple[str, ...] | None
end_user_metadata: Mapping[str, dict[str, object]]
async def resolve_customer_daily_activity_scope(
*,
end_user_ids: tuple[str, ...] | None,
prisma_client: "PrismaClient",
) -> _CustomerDailyActivityScope:
end_user_table: Final = _typed_table(EndUserRepository(prisma_client))
end_user_aliases: Final = (
await find_many_in(end_user_table, "user_id", end_user_ids)
if end_user_ids is not None
else await end_user_table.find_many(where={})
)
metadata: Final = MappingProxyType({end_user.user_id: {"alias": end_user.alias} for end_user in end_user_aliases})
return _CustomerDailyActivityScope(end_user_ids, metadata)
def customer_daily_activity_is_admin(user_api_key_dict: UserAPIKeyAuth) -> bool:
return user_api_key_dict.user_role in (
LitellmUserRoles.PROXY_ADMIN,
LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY,
)
@router.get(
"/customer/info",
tags=["Customer Management"],
@ -888,10 +917,7 @@ async def get_customer_daily_activity(
"""
Get daily activity for specific organizations or all accessible organizations.
"""
if (
user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN
and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY
):
if not customer_daily_activity_is_admin(user_api_key_dict):
raise HTTPException(
status_code=401,
detail={"error": f"Admin-only endpoint. Your user role={user_api_key_dict.user_role}"},
@ -906,24 +932,22 @@ async def get_customer_daily_activity(
)
# Parse comma-separated ids
end_user_ids_list: Final = end_user_ids.split(",") if end_user_ids else None
end_user_ids_list: Final = tuple(end_user_ids.split(",")) if end_user_ids else None
exclude_end_user_ids_list: list[str] | None = None
if exclude_end_user_ids:
exclude_end_user_ids_list = exclude_end_user_ids.split(",") if exclude_end_user_ids else None
# Fetch organization aliases for metadata
where_condition: Final = dict[str, object]()
if end_user_ids_list:
where_condition["user_id"] = {"in": list(end_user_ids_list)}
end_user_aliases: Final = await _typed_table(EndUserRepository(prisma_client)).find_many(where=where_condition)
customer_scope: Final = await resolve_customer_daily_activity_scope(
end_user_ids=end_user_ids_list,
prisma_client=prisma_client,
)
# Query daily activity for organizations
return await get_daily_activity(
prisma_client=prisma_client,
table_name="litellm_dailyenduserspend",
entity_id_field="end_user_id",
entity_id=end_user_ids_list,
entity_metadata_field={e.user_id: {"alias": e.alias} for e in end_user_aliases},
entity_id=None if customer_scope.end_user_ids is None else list(customer_scope.end_user_ids),
entity_metadata_field=customer_scope.end_user_metadata,
exclude_entity_ids=exclude_end_user_ids_list,
start_date=start_date,
end_date=end_date,

579
litellm/proxy/management_endpoints/daily_activity_routes.py Normal file
View file

@ -0,0 +1,579 @@
import csv
import io
import json
from collections.abc import AsyncIterator, Mapping, Sequence
from dataclasses import asdict, fields, replace
from datetime import datetime
from typing import Annotated, Final, Literal
from fastapi import APIRouter, Depends, HTTPException, Query
from fastapi.encoders import jsonable_encoder
from fastapi.responses import StreamingResponse
from litellm import constants
from litellm._logging import verbose_proxy_logger
from litellm.proxy._types import CommonProxyErrors, UserAPIKeyAuth
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.management_endpoints.common_daily_activity import (
InvalidDateRange,
ScopeDenied,
daily_activity_repository,
get_daily_activity_aggregated,
raise_public,
spend_logs_window,
)
from litellm.proxy.management_endpoints.daily_activity_scopes import (
AGENT_RESOLVER,
CUSTOMER_RESOLVER,
ORGANIZATION_RESOLVER,
TAG_RESOLVER,
TEAM_RESOLVER,
USER_RESOLVER,
EntityQuery,
EntityScopeResolver,
ResolvedScope,
)
from litellm.proxy.management_endpoints.team_endpoints import aggregated_date_range_error
from litellm.proxy.management_helpers.utils import management_endpoint_wrapper
from litellm.proxy.utils import PrismaClient, get_prisma_client_or_throw
from litellm.repositories.daily_activity_repository import DailyActivityRepository
from litellm.types.proxy.management_endpoints.common_daily_activity import (
CacheLeakageKeysResponse,
DailyActivityKeyPageResponse,
DailyActivityKeySearchResponse,
KeyActivityRow,
KeyMetadata,
KeySpendActivityRow,
KeySpendMetrics,
ModelTopKeysResponse,
SpendAnalyticsPaginatedResponse,
SpendMetrics,
)
from litellm.types.repositories.daily_activity import (
ExportRow,
ExportType,
KeyMetadataRow,
KeySpendRow,
)
router = APIRouter()
def get_daily_activity_prisma_client() -> PrismaClient:
return get_prisma_client_or_throw(CommonProxyErrors.db_not_connected_error.value)
def get_daily_activity_repository() -> DailyActivityRepository:
return daily_activity_repository(get_daily_activity_prisma_client())
def _date_range_error(query: EntityQuery, *, user_aggregated: bool) -> InvalidDateRange | None:
if user_aggregated:
if query.start_date is None or query.end_date is None:
return InvalidDateRange(reason="Please provide start_date and end_date")
return None
range_error: Final[str | None] = aggregated_date_range_error(query.start_date, query.end_date)
return None if range_error is None else InvalidDateRange(reason=range_error)
async def _resolved_scope(
resolver: EntityScopeResolver,
query: EntityQuery,
user_api_key_dict: UserAPIKeyAuth,
prisma_client: PrismaClient,
*,
user_aggregated: bool,
) -> ResolvedScope:
date_error: Final[InvalidDateRange | None] = _date_range_error(query, user_aggregated=user_aggregated)
if date_error is not None:
raise_public(date_error)
result: ResolvedScope | ScopeDenied = await resolver.resolve(user_api_key_dict, query, prisma_client)
if isinstance(result, ScopeDenied):
raise_public(result)
return result
def _sum_metrics(metrics: Sequence[SpendMetrics]) -> SpendMetrics:
return SpendMetrics(
spend=sum(metric.spend for metric in metrics),
flat_cost=sum(metric.flat_cost for metric in metrics),
prompt_tokens=sum(metric.prompt_tokens for metric in metrics),
completion_tokens=sum(metric.completion_tokens for metric in metrics),
cache_read_input_tokens=sum(metric.cache_read_input_tokens for metric in metrics),
cache_creation_input_tokens=sum(metric.cache_creation_input_tokens for metric in metrics),
compression_saved_tokens=sum(metric.compression_saved_tokens for metric in metrics),
compression_savings_spend=sum(metric.compression_savings_spend for metric in metrics),
prompt_caching_savings_spend=sum(metric.prompt_caching_savings_spend for metric in metrics),
gateway_injected_caching_savings_spend=sum(metric.gateway_injected_caching_savings_spend for metric in metrics),
autorouter_savings_spend=sum(metric.autorouter_savings_spend for metric in metrics),
total_tokens=sum(metric.total_tokens for metric in metrics),
successful_requests=sum(metric.successful_requests for metric in metrics),
failed_requests=sum(metric.failed_requests for metric in metrics),
api_requests=sum(metric.api_requests for metric in metrics),
total_response_time_ms=sum(metric.total_response_time_ms for metric in metrics),
timed_requests=sum(metric.timed_requests for metric in metrics),
)
def _key_metadata(api_key: str, metadata: Mapping[str, KeyMetadataRow]) -> KeyMetadata:
row: Final[KeyMetadataRow | None] = metadata.get(api_key)
if row is None:
return KeyMetadata()
return KeyMetadata(
key_alias=row.key_alias,
team_id=row.team_id,
user_id=row.user_id,
user_email=row.user_email,
key_exists=row.key_exists,
)
def _key_activity_row(row: KeySpendRow, metadata: Mapping[str, KeyMetadataRow]) -> KeySpendActivityRow:
return KeySpendActivityRow(
api_key=row.api_key,
metrics=KeySpendMetrics(
spend=row.spend,
prompt_tokens=row.prompt_tokens,
completion_tokens=row.completion_tokens,
total_tokens=row.total_tokens,
api_requests=row.api_requests,
successful_requests=row.successful_requests,
failed_requests=row.failed_requests,
cache_read_input_tokens=row.cache_read_input_tokens,
cache_creation_input_tokens=row.cache_creation_input_tokens,
),
metadata=_key_metadata(row.api_key, metadata),
)
async def _key_activity_rows(
repository: DailyActivityRepository,
rows: Sequence[KeySpendRow],
resolved_scope: ResolvedScope,
) -> list[KeySpendActivityRow]:
spend_window: Final[tuple[datetime, datetime] | None] = spend_logs_window(
frozenset((resolved_scope.scope.start_date, resolved_scope.scope.end_date))
)
metadata: Final[Mapping[str, KeyMetadataRow]] = await repository.key_metadata(
frozenset(row.api_key for row in rows),
spend_window,
)
return [_key_activity_row(row, metadata) for row in rows]
def _export_filename(
entity: str,
start_date: str,
end_date: str,
export_type: ExportType,
file_format: Literal["csv", "json"],
) -> str:
extension: Final[str] = "csv" if file_format == "csv" else "json"
return f"{entity}-usage-{start_date}-{end_date}-{export_type.value}.{extension}"
def _content_disposition(
entity: str,
start_date: str,
end_date: str,
export_type: ExportType,
file_format: Literal["csv", "json"],
) -> str:
filename: Final = _export_filename(entity, start_date, end_date, export_type, file_format)
return f'attachment; filename="{filename}"'
def _fold_key_metrics(api_key: str, response: SpendAnalyticsPaginatedResponse) -> KeyActivityRow | None:
metrics: Final[tuple[SpendMetrics, ...]] = tuple(
day.breakdown.api_keys[api_key].metrics for day in response.results if api_key in day.breakdown.api_keys
)
if not metrics:
return None
metadata: Final[KeyMetadata] = next(
day.breakdown.api_keys[api_key].metadata for day in response.results if api_key in day.breakdown.api_keys
)
return KeyActivityRow(api_key=api_key, metrics=_sum_metrics(metrics), metadata=metadata)
def _search_rows(keys: Sequence[str], response: SpendAnalyticsPaginatedResponse) -> list[KeyActivityRow]:
return [row for key in keys if (row := _fold_key_metrics(key, response)) is not None]
def _csv_cell(value: object) -> object:
if isinstance(value, str) and value.startswith(("=", "+", "-", "@", "\t", "\r")):
return f"'{value}"
return value
def _csv_row(values: Sequence[object]) -> bytes:
output: Final[io.StringIO] = io.StringIO(newline="")
csv.writer(output, lineterminator="\r\n").writerow(tuple(_csv_cell(value) for value in values))
return output.getvalue().encode()
def _stream_export_rows(
first_row: ExportRow | None,
rows: AsyncIterator[ExportRow],
file_format: Literal["csv", "json"],
) -> AsyncIterator[bytes]:
async def stream() -> AsyncIterator[bytes]:
if file_format == "csv":
yield _csv_row(tuple(field.name for field in fields(ExportRow)))
if first_row is not None:
yield _csv_row(tuple(asdict(first_row).values()))
async for row in rows:
yield _csv_row(tuple(asdict(row).values()))
return
if first_row is None:
yield b"[]"
return
yield b"[" + json.dumps(jsonable_encoder(first_row), separators=(",", ":")).encode()
async for row in rows:
yield b"," + json.dumps(jsonable_encoder(row), separators=(",", ":")).encode()
yield b"]"
return stream()
def _register_aggregated_route(router: APIRouter, resolver: EntityScopeResolver, prefix: str) -> None:
@management_endpoint_wrapper
async def aggregated(
entity_query: Annotated[EntityQuery, Depends(resolver.query)],
user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
repository: Annotated[DailyActivityRepository, Depends(get_daily_activity_repository)],
prisma_client: Annotated[PrismaClient, Depends(get_daily_activity_prisma_client)],
api_key_limit: Annotated[
int, Query(ge=1, le=constants.USAGE_TOP_API_KEYS_MAX)
] = constants.USAGE_TOP_API_KEYS_DEFAULT,
) -> SpendAnalyticsPaginatedResponse:
try:
resolved: ResolvedScope = await _resolved_scope(
resolver,
entity_query,
user_api_key_dict,
prisma_client,
user_aggregated=resolver.entity == "user",
)
return await get_daily_activity_aggregated(
repository,
resolved.scope,
entity_metadata_field=resolved.entity_metadata,
include_entity_breakdown=resolver.include_entity_breakdown,
api_key_limit=api_key_limit,
)
except HTTPException:
raise
except Exception as exc:
verbose_proxy_logger.exception("Daily activity aggregation failed: %s", exc)
raise HTTPException(status_code=500, detail={"error": f"Failed to fetch analytics: {exc}"})
router.add_api_route(
f"{prefix}/daily/activity/aggregated",
aggregated,
methods=["GET"],
name=resolver.operation_names["aggregated"],
tags=list(resolver.tags),
dependencies=(Depends(user_api_key_auth),),
response_model=SpendAnalyticsPaginatedResponse,
include_in_schema=prefix != "/end_user",
)
if resolver.entity == "user":
aggregated.__doc__ = (
"Aggregated analytics for a user's daily activity without pagination.\n"
"Returns the same response shape as the paginated endpoint with page metadata set to single-page.\n\n"
"Reads daily spend records that only ever accumulate and are never affected by budget\n"
"resets. Their total can legitimately exceed the `spend` field returned by\n"
"`/v2/user/info`, which is a running budget counter that every budget reset sets back\n"
"to zero (or to the overage above `max_budget` when `budget_rollover` is enabled)."
)
elif resolver.entity == "team":
aggregated.__doc__ = (
"Aggregated daily activity for teams without pagination, including per-team breakdown.\n\n"
"One SQL GROUPING SETS pass returns every day in the range regardless of row\n"
"volume, so callers never reassemble pages. Same response shape as the\n"
"paginated endpoint with page metadata pinned to a single page.\n\n"
"Args:\n"
" team_ids (Optional[str]): Comma-separated list of team IDs to filter by. If not provided, "
"returns data for all teams.\n"
" start_date (Optional[str]): Start date for the activity period (YYYY-MM-DD).\n"
" end_date (Optional[str]): End date for the activity period (YYYY-MM-DD).\n"
" model (Optional[str]): Filter by model name.\n"
" api_key (Optional[str]): Filter by API key.\n"
" exclude_team_ids (Optional[str]): Comma-separated list of team IDs to exclude.\n"
" timezone (Optional[int]): Timezone offset in minutes from UTC, matching JavaScript's "
"Date.getTimezoneOffset() convention.\n"
"Returns:\n"
" SpendAnalyticsPaginatedResponse: Response containing all daily activity data for the range."
)
def _register_key_page_route(router: APIRouter, resolver: EntityScopeResolver, prefix: str) -> None:
@management_endpoint_wrapper
async def key_page(
entity_query: Annotated[EntityQuery, Depends(resolver.query)],
user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
repository: Annotated[DailyActivityRepository, Depends(get_daily_activity_repository)],
prisma_client: Annotated[PrismaClient, Depends(get_daily_activity_prisma_client)],
offset: Annotated[int, Query(ge=0)] = 0,
limit: Annotated[int, Query(ge=1, le=constants.USAGE_KEY_PAGE_MAX)] = constants.USAGE_KEY_PAGE_DEFAULT,
) -> DailyActivityKeyPageResponse:
try:
resolved: Final = await _resolved_scope(
resolver,
entity_query,
user_api_key_dict,
prisma_client,
user_aggregated=False,
)
page: Final = await repository.key_page(resolved.scope, offset=offset, limit=limit)
return DailyActivityKeyPageResponse(
api_keys=await _key_activity_rows(repository, page.rows, resolved),
total_api_keys=page.total_api_keys,
offset=offset,
limit=limit,
)
except HTTPException:
raise
except Exception as exc:
verbose_proxy_logger.exception("Daily activity key page failed: %s", exc)
raise HTTPException(status_code=500, detail={"error": f"Failed to fetch analytics: {exc}"})
router.add_api_route(
f"{prefix}/daily/activity/aggregated/keys",
key_page,
methods=["GET"],
name=resolver.operation_names["key_page"],
tags=list(resolver.tags),
dependencies=(Depends(user_api_key_auth),),
response_model=DailyActivityKeyPageResponse,
include_in_schema=prefix != "/end_user",
)
def _register_search_route(router: APIRouter, resolver: EntityScopeResolver, prefix: str) -> None:
@management_endpoint_wrapper
async def search(
entity_query: Annotated[EntityQuery, Depends(resolver.query)],
search: Annotated[str, Query(min_length=1)],
user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
repository: Annotated[DailyActivityRepository, Depends(get_daily_activity_repository)],
prisma_client: Annotated[PrismaClient, Depends(get_daily_activity_prisma_client)],
limit: Annotated[int, Query(ge=1, le=constants.USAGE_KEY_SEARCH_MAX)] = constants.USAGE_KEY_SEARCH_DEFAULT,
) -> DailyActivityKeySearchResponse:
try:
resolved: ResolvedScope = await _resolved_scope(
resolver,
entity_query,
user_api_key_dict,
prisma_client,
user_aggregated=False,
)
keys: tuple[str, ...] = await repository.search_keys(
resolved.scope,
search=search,
limit=limit,
)
if not keys:
return DailyActivityKeySearchResponse(api_keys=[])
search_response: SpendAnalyticsPaginatedResponse = await get_daily_activity_aggregated(
repository,
replace(resolved.scope, api_keys=keys),
include_entity_breakdown=False,
)
return DailyActivityKeySearchResponse(api_keys=_search_rows(keys, search_response))
except HTTPException:
raise
except Exception as exc:
verbose_proxy_logger.exception("Daily activity key search failed: %s", exc)
raise HTTPException(status_code=500, detail={"error": f"Failed to fetch analytics: {exc}"})
router.add_api_route(
f"{prefix}/daily/activity/aggregated/search",
search,
methods=["GET"],
name=resolver.operation_names["search"],
tags=list(resolver.tags),
dependencies=(Depends(user_api_key_auth),),
response_model=DailyActivityKeySearchResponse,
include_in_schema=prefix != "/end_user",
)
def _register_model_top_keys_route(router: APIRouter, resolver: EntityScopeResolver, prefix: str) -> None:
@management_endpoint_wrapper
async def model_top_keys(
entity_query: Annotated[EntityQuery, Depends(resolver.query)],
user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
repository: Annotated[DailyActivityRepository, Depends(get_daily_activity_repository)],
prisma_client: Annotated[PrismaClient, Depends(get_daily_activity_prisma_client)],
model_group: Annotated[str, Query(min_length=1)],
by_model_group: Annotated[bool, Query()] = True,
limit: Annotated[int, Query(ge=1, le=constants.USAGE_MODEL_TOP_KEYS_MAX)] = (
constants.USAGE_MODEL_TOP_KEYS_DEFAULT
),
) -> ModelTopKeysResponse:
try:
resolved: ResolvedScope = await _resolved_scope(
resolver,
entity_query,
user_api_key_dict,
prisma_client,
user_aggregated=False,
)
rows: tuple[KeySpendRow, ...] = await repository.model_top_keys(
resolved.scope,
model_group=model_group,
by_model_group=by_model_group,
limit=limit,
)
return ModelTopKeysResponse(
model=model_group,
by_model_group=by_model_group,
api_keys=await _key_activity_rows(repository, rows, resolved),
)
except HTTPException:
raise
except Exception as exc:
verbose_proxy_logger.exception("Daily activity model top keys failed: %s", exc)
raise HTTPException(status_code=500, detail={"error": f"Failed to fetch analytics: {exc}"})
router.add_api_route(
f"{prefix}/daily/activity/aggregated/model_top_keys",
model_top_keys,
methods=["GET"],
name=resolver.operation_names["model_top_keys"],
tags=list(resolver.tags),
dependencies=(Depends(user_api_key_auth),),
response_model=ModelTopKeysResponse,
include_in_schema=prefix != "/end_user",
)
def _register_export_route(router: APIRouter, resolver: EntityScopeResolver, prefix: str) -> None:
@management_endpoint_wrapper
async def export(
entity_query: Annotated[EntityQuery, Depends(resolver.query)],
user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
repository: Annotated[DailyActivityRepository, Depends(get_daily_activity_repository)],
prisma_client: Annotated[PrismaClient, Depends(get_daily_activity_prisma_client)],
export_type: Annotated[ExportType, Query()] = ExportType.DAILY,
file_format: Annotated[Literal["csv", "json"], Query(alias="format")] = "csv",
) -> StreamingResponse:
try:
resolved: ResolvedScope = await _resolved_scope(
resolver,
entity_query,
user_api_key_dict,
prisma_client,
user_aggregated=False,
)
rows: Final = repository.export_rows(resolved.scope, export_type=export_type)
first_row: Final = await anext(rows, None)
return StreamingResponse(
_stream_export_rows(first_row, rows, file_format),
media_type="text/csv" if file_format == "csv" else "application/json",
headers={
"Cache-Control": "no-store",
"Content-Disposition": _content_disposition(
resolver.entity,
resolved.scope.start_date,
resolved.scope.end_date,
export_type,
file_format,
),
},
)
except HTTPException:
raise
except Exception as exc:
verbose_proxy_logger.exception("Daily activity export failed: %s", exc)
raise HTTPException(status_code=500, detail={"error": f"Failed to fetch analytics: {exc}"})
router.add_api_route(
f"{prefix}/daily/activity/export",
export,
methods=["GET"],
name=resolver.operation_names["export"],
tags=list(resolver.tags),
dependencies=(Depends(user_api_key_auth),),
response_class=StreamingResponse,
responses={
200: {
"description": "Streamed daily activity export",
"content": {
"text/csv": {"schema": {"type": "string"}},
"application/json": {"schema": {"type": "array", "items": {"type": "object"}}},
},
}
},
include_in_schema=prefix != "/end_user",
)
def _register_cache_leakage_route(router: APIRouter, resolver: EntityScopeResolver, prefix: str) -> None:
@management_endpoint_wrapper
async def cache_leakage_keys(
entity_query: Annotated[EntityQuery, Depends(resolver.query)],
user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
repository: Annotated[DailyActivityRepository, Depends(get_daily_activity_repository)],
prisma_client: Annotated[PrismaClient, Depends(get_daily_activity_prisma_client)],
limit: Annotated[
int, Query(ge=1, le=constants.USAGE_CACHE_LEAKAGE_KEYS_MAX)
] = constants.USAGE_CACHE_LEAKAGE_KEYS_DEFAULT,
) -> CacheLeakageKeysResponse:
try:
resolved: ResolvedScope = await _resolved_scope(
resolver,
entity_query,
user_api_key_dict,
prisma_client,
user_aggregated=False,
)
rows: tuple[KeySpendRow, ...] = await repository.cache_leakage_keys(
resolved.scope,
limit=limit,
)
return CacheLeakageKeysResponse(
api_keys=await _key_activity_rows(repository, rows, resolved),
)
except HTTPException:
raise
except Exception as exc:
verbose_proxy_logger.exception("Daily activity cache leakage keys failed: %s", exc)
raise HTTPException(status_code=500, detail={"error": f"Failed to fetch analytics: {exc}"})
router.add_api_route(
f"{prefix}/daily/activity/aggregated/cache_leakage_keys",
cache_leakage_keys,
methods=["GET"],
name=resolver.operation_names["cache_leakage_keys"],
tags=list(resolver.tags),
dependencies=(Depends(user_api_key_auth),),
response_model=CacheLeakageKeysResponse,
include_in_schema=prefix != "/end_user",
)
def register_daily_activity_routes(router: APIRouter, resolver: EntityScopeResolver) -> None:
for prefix in resolver.route_prefixes:
_register_aggregated_route(router, resolver, prefix)
_register_key_page_route(router, resolver, prefix)
_register_search_route(router, resolver, prefix)
_register_model_top_keys_route(router, resolver, prefix)
_register_export_route(router, resolver, prefix)
if resolver.entity == "user":
_register_cache_leakage_route(router, resolver, prefix)
for _resolver in (
USER_RESOLVER,
TEAM_RESOLVER,
TAG_RESOLVER,
ORGANIZATION_RESOLVER,
CUSTOMER_RESOLVER,
AGENT_RESOLVER,
):
register_daily_activity_routes(router, _resolver)

472
litellm/proxy/management_endpoints/daily_activity_scopes.py Normal file
View file

@ -0,0 +1,472 @@
from collections.abc import Awaitable, Callable, Mapping, Sequence
from dataclasses import dataclass
from types import MappingProxyType
from typing import Final, Literal
from fastapi import Query
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.agent_endpoints.endpoints import resolve_agent_daily_activity_scope
from litellm.proxy.management_endpoints.common_daily_activity import ScopeDenied
from litellm.proxy.management_endpoints.customer_endpoints import (
customer_daily_activity_is_admin,
resolve_customer_daily_activity_scope,
)
from litellm.proxy.management_endpoints.internal_user_endpoints import resolve_user_daily_activity_entity_ids
from litellm.proxy.management_endpoints.organization_endpoints import resolve_organization_daily_activity_scope
from litellm.proxy.management_endpoints.tag_management_endpoints import get_tag_daily_activity_api_key_filter
from litellm.proxy.management_endpoints.team_endpoints import resolve_team_daily_activity_scope
from litellm.proxy.utils import PrismaClient
from litellm.types.repositories.daily_activity import DailyActivityScope, DailyActivityTable
@dataclass(frozen=True, slots=True)
class EntityQuery:
entity_ids: tuple[str, ...] | None
exclude_entity_ids: tuple[str, ...]
api_key: str | None
start_date: str | None
end_date: str | None
model: str | None
timezone_offset_minutes: int | None
include_current_utc_day: bool
@dataclass(frozen=True, slots=True)
class ResolvedScope:
scope: DailyActivityScope
entity_metadata: Mapping[str, dict[str, object]] | None
Entity = Literal["user", "team", "tag", "organization", "customer", "agent"]
EntityScopeResolution = ResolvedScope | ScopeDenied
EntityScopeQuery = Callable[..., EntityQuery]
EntityScopeResolve = Callable[[UserAPIKeyAuth, EntityQuery, PrismaClient], Awaitable[EntityScopeResolution]]
OperationNames = Mapping[str, str]
@dataclass(frozen=True, slots=True)
class EntityScopeResolver:
entity: Entity
table: DailyActivityTable
entity_id_field: str
route_prefixes: tuple[str, ...]
tags: tuple[str, ...]
query: EntityScopeQuery
resolve: EntityScopeResolve
include_entity_breakdown: bool
operation_names: OperationNames
def _query_ids(value: str | None) -> tuple[str, ...] | None:
return tuple(value.split(",")) if value else None
def _query_excluded_ids(value: str | None) -> tuple[str, ...]:
return tuple(value.split(",")) if value else ()
def _build_scope(
resolver: EntityScopeResolver,
query: EntityQuery,
entity_ids: Sequence[str] | None,
exclude_entity_ids: Sequence[str],
api_key_filter: str | Sequence[str] | None,
entity_metadata: Mapping[str, dict[str, object]] | None,
) -> ResolvedScope:
start_date: Final[str] = query.start_date or ""
end_date: Final[str] = query.end_date or ""
api_keys: Final[tuple[str, ...] | None] = (
None
if api_key_filter is None or api_key_filter == ""
else (api_key_filter,)
if isinstance(api_key_filter, str)
else tuple(api_key_filter)
)
return ResolvedScope(
scope=DailyActivityScope(
table=resolver.table,
entity_id_field=resolver.entity_id_field,
entity_ids=None if entity_ids is None else tuple(entity_ids),
exclude_entity_ids=tuple(exclude_entity_ids),
api_keys=api_keys,
start_date=start_date,
end_date=end_date,
model=query.model,
timezone_offset_minutes=query.timezone_offset_minutes,
include_current_utc_day=query.include_current_utc_day,
),
entity_metadata=entity_metadata,
)
async def _resolve_user(
user_api_key_dict: UserAPIKeyAuth, query: EntityQuery, prisma_client: PrismaClient
) -> EntityScopeResolution:
entity_ids: Final = resolve_user_daily_activity_entity_ids(
user_id=query.entity_ids[0] if query.entity_ids is not None else None,
user_api_key_dict=user_api_key_dict,
)
if isinstance(entity_ids, ScopeDenied):
return entity_ids
return _build_scope(
USER_RESOLVER,
query,
entity_ids,
query.exclude_entity_ids,
query.api_key,
None,
)
async def _resolve_team(
user_api_key_dict: UserAPIKeyAuth, query: EntityQuery, prisma_client: PrismaClient
) -> EntityScopeResolution:
from litellm.proxy.proxy_server import proxy_logging_obj, user_api_key_cache
team_scope: Final = await resolve_team_daily_activity_scope(
team_ids=",".join(query.entity_ids) if query.entity_ids is not None else None,
exclude_team_ids=",".join(query.exclude_entity_ids) if query.exclude_entity_ids else None,
api_key=query.api_key,
user_api_key_dict=user_api_key_dict,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
proxy_logging_obj=proxy_logging_obj,
)
return _build_scope(
TEAM_RESOLVER,
query,
team_scope.team_ids,
team_scope.exclude_team_ids or (),
team_scope.api_key_filter,
team_scope.team_alias_metadata,
)
async def _resolve_tag(
user_api_key_dict: UserAPIKeyAuth, query: EntityQuery, prisma_client: PrismaClient
) -> EntityScopeResolution:
api_key_filter: Final = await get_tag_daily_activity_api_key_filter(
prisma_client=prisma_client,
user_api_key_dict=user_api_key_dict,
requested_api_key=query.api_key,
)
return _build_scope(
TAG_RESOLVER,
query,
query.entity_ids,
query.exclude_entity_ids,
api_key_filter,
None,
)
async def _resolve_organization(
user_api_key_dict: UserAPIKeyAuth, query: EntityQuery, prisma_client: PrismaClient
) -> EntityScopeResolution:
org_scope: Final = await resolve_organization_daily_activity_scope(
organization_ids=query.entity_ids,
prisma_client=prisma_client,
user_api_key_dict=user_api_key_dict,
)
return _build_scope(
ORGANIZATION_RESOLVER,
query,
org_scope.organization_ids,
query.exclude_entity_ids,
query.api_key,
org_scope.organization_metadata,
)
async def _resolve_customer(
user_api_key_dict: UserAPIKeyAuth, query: EntityQuery, prisma_client: PrismaClient
) -> EntityScopeResolution:
if not customer_daily_activity_is_admin(user_api_key_dict):
return ScopeDenied(403, f"Admin-only endpoint. Your user role={user_api_key_dict.user_role}")
customer_scope: Final = await resolve_customer_daily_activity_scope(
end_user_ids=query.entity_ids,
prisma_client=prisma_client,
)
return _build_scope(
CUSTOMER_RESOLVER,
query,
customer_scope.end_user_ids,
query.exclude_entity_ids,
query.api_key,
customer_scope.end_user_metadata,
)
async def _resolve_agent(
user_api_key_dict: UserAPIKeyAuth, query: EntityQuery, prisma_client: PrismaClient
) -> EntityScopeResolution:
agent_scope: Final = await resolve_agent_daily_activity_scope(
agent_ids=query.entity_ids,
user_api_key_dict=user_api_key_dict,
prisma_client=prisma_client,
)
return _build_scope(
AGENT_RESOLVER,
query,
agent_scope.agent_ids,
query.exclude_entity_ids,
query.api_key,
agent_scope.agent_metadata,
)
def _user_query(
start_date: str | None = Query(default=None, description="Start date in YYYY-MM-DD format"),
end_date: str | None = Query(default=None, description="End date in YYYY-MM-DD format"),
model: str | None = Query(default=None, description="Filter by specific model"),
api_key: str | None = Query(default=None, description="Filter by specific API key"),
user_id: str | None = Query(
default=None,
description="Filter by specific user ID. Admins can filter by any user or omit for global view. "
"Non-admins must provide their own user_id.",
),
timezone: int | None = Query(
default=None,
description="Timezone offset in minutes from UTC (e.g., 480 for PST). "
"Matches JavaScript's Date.getTimezoneOffset() convention.",
),
include_current_utc_day: bool = Query(
default=False,
description="When the range ends on the caller's current local day, extend it to "
"today's UTC bucket so spend written after the caller's local midnight (in UTC "
"terms) is included. Requires the timezone parameter. Historical ranges are "
"never extended.",
),
) -> EntityQuery:
return EntityQuery(
entity_ids=(user_id,) if user_id is not None else None,
exclude_entity_ids=(),
api_key=api_key,
start_date=start_date,
end_date=end_date,
model=model,
timezone_offset_minutes=timezone,
include_current_utc_day=include_current_utc_day,
)
def _team_query(
team_ids: str | None = None,
start_date: str | None = None,
end_date: str | None = None,
model: str | None = None,
api_key: str | None = None,
exclude_team_ids: str | None = None,
timezone: int | None = None,
) -> EntityQuery:
return EntityQuery(
entity_ids=_query_ids(team_ids),
exclude_entity_ids=_query_excluded_ids(exclude_team_ids),
api_key=api_key,
start_date=start_date,
end_date=end_date,
model=model,
timezone_offset_minutes=timezone,
include_current_utc_day=False,
)
def _tag_query(
start_date: str | None = None,
end_date: str | None = None,
model: str | None = None,
api_key: str | None = None,
tags: str | None = None,
timezone: int | None = None,
) -> EntityQuery:
return EntityQuery(
entity_ids=_query_ids(tags),
exclude_entity_ids=(),
api_key=api_key,
start_date=start_date,
end_date=end_date,
model=model,
timezone_offset_minutes=timezone,
include_current_utc_day=False,
)
def _organization_query(
organization_ids: str | None = None,
start_date: str | None = None,
end_date: str | None = None,
model: str | None = None,
api_key: str | None = None,
exclude_organization_ids: str | None = None,
timezone: int | None = None,
) -> EntityQuery:
return EntityQuery(
entity_ids=_query_ids(organization_ids),
exclude_entity_ids=_query_excluded_ids(exclude_organization_ids),
api_key=api_key,
start_date=start_date,
end_date=end_date,
model=model,
timezone_offset_minutes=timezone,
include_current_utc_day=False,
)
def _customer_query(
end_user_ids: str | None = None,
start_date: str | None = None,
end_date: str | None = None,
model: str | None = None,
api_key: str | None = None,
exclude_end_user_ids: str | None = None,
timezone: int | None = None,
) -> EntityQuery:
return EntityQuery(
entity_ids=_query_ids(end_user_ids),
exclude_entity_ids=_query_excluded_ids(exclude_end_user_ids),
api_key=api_key,
start_date=start_date,
end_date=end_date,
model=model,
timezone_offset_minutes=timezone,
include_current_utc_day=False,
)
def _agent_query(
agent_ids: str | None = None,
start_date: str | None = None,
end_date: str | None = None,
model: str | None = None,
api_key: str | None = None,
exclude_agent_ids: str | None = None,
timezone: int | None = None,
) -> EntityQuery:
return EntityQuery(
entity_ids=_query_ids(agent_ids),
exclude_entity_ids=_query_excluded_ids(exclude_agent_ids),
api_key=api_key,
start_date=start_date,
end_date=end_date,
model=model,
timezone_offset_minutes=timezone,
include_current_utc_day=False,
)
USER_RESOLVER = EntityScopeResolver(
entity="user",
table=DailyActivityTable.USER,
entity_id_field="user_id",
route_prefixes=("/user",),
tags=("Budget & Spend Tracking", "Internal User management"),
query=_user_query,
resolve=_resolve_user,
include_entity_breakdown=False,
operation_names=MappingProxyType(
{
"aggregated": "get_user_daily_activity_aggregated",
"search": "get_user_daily_activity_aggregated_search",
"key_page": "get_user_daily_activity_aggregated_keys",
"model_top_keys": "get_user_daily_activity_model_top_keys",
"export": "get_user_daily_activity_export",
"cache_leakage_keys": "get_user_daily_activity_cache_leakage_keys",
}
),
)
TEAM_RESOLVER = EntityScopeResolver(
entity="team",
table=DailyActivityTable.TEAM,
entity_id_field="team_id",
route_prefixes=("/team",),
tags=("team management",),
query=_team_query,
resolve=_resolve_team,
include_entity_breakdown=True,
operation_names=MappingProxyType(
{
"aggregated": "get_team_daily_activity_aggregated",
"search": "get_team_daily_activity_aggregated_search",
"key_page": "get_team_daily_activity_aggregated_keys",
"model_top_keys": "get_team_daily_activity_model_top_keys",
"export": "get_team_daily_activity_export",
}
),
)
TAG_RESOLVER = EntityScopeResolver(
entity="tag",
table=DailyActivityTable.TAG,
entity_id_field="tag",
route_prefixes=("/tag",),
tags=("tag management",),
query=_tag_query,
resolve=_resolve_tag,
include_entity_breakdown=True,
operation_names=MappingProxyType(
{
"aggregated": "get_tag_daily_activity_aggregated",
"search": "get_tag_daily_activity_aggregated_search",
"key_page": "get_tag_daily_activity_aggregated_keys",
"model_top_keys": "get_tag_daily_activity_model_top_keys",
"export": "get_tag_daily_activity_export",
}
),
)
ORGANIZATION_RESOLVER = EntityScopeResolver(
entity="organization",
table=DailyActivityTable.ORGANIZATION,
entity_id_field="organization_id",
route_prefixes=("/organization",),
tags=("organization management",),
query=_organization_query,
resolve=_resolve_organization,
include_entity_breakdown=True,
operation_names=MappingProxyType(
{
"aggregated": "get_organization_daily_activity_aggregated",
"search": "get_organization_daily_activity_aggregated_search",
"key_page": "get_organization_daily_activity_aggregated_keys",
"model_top_keys": "get_organization_daily_activity_model_top_keys",
"export": "get_organization_daily_activity_export",
}
),
)
CUSTOMER_RESOLVER = EntityScopeResolver(
entity="customer",
table=DailyActivityTable.CUSTOMER,
entity_id_field="end_user_id",
route_prefixes=("/customer", "/end_user"),
tags=("Customer Management",),
query=_customer_query,
resolve=_resolve_customer,
include_entity_breakdown=True,
operation_names=MappingProxyType(
{
"aggregated": "get_customer_daily_activity_aggregated",
"search": "get_customer_daily_activity_aggregated_search",
"key_page": "get_customer_daily_activity_aggregated_keys",
"model_top_keys": "get_customer_daily_activity_model_top_keys",
"export": "get_customer_daily_activity_export",
}
),
)
AGENT_RESOLVER = EntityScopeResolver(
entity="agent",
table=DailyActivityTable.AGENT,
entity_id_field="agent_id",
route_prefixes=("/agent",),
tags=("Agent Management",),
query=_agent_query,
resolve=_resolve_agent,
include_entity_breakdown=True,
operation_names=MappingProxyType(
{
"aggregated": "get_agent_daily_activity_aggregated",
"search": "get_agent_daily_activity_aggregated_search",
"key_page": "get_agent_daily_activity_aggregated_keys",
"model_top_keys": "get_agent_daily_activity_model_top_keys",
"export": "get_agent_daily_activity_export",
}
),
)

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

@ -54,10 +54,9 @@ from litellm.proxy.hooks.user_management_event_hooks import UserManagementEventH
from litellm.proxy.management.teams.access import is_team_admin
from litellm.proxy.management_endpoints.common_daily_activity import (
DailySpendRecord,
daily_activity_repository,
daily_activity_scope,
ScopeDenied,
get_daily_activity,
get_daily_activity_aggregated,
raise_public,
)
from litellm.proxy.management_endpoints.common_utils import (
_user_has_admin_view,
@ -2864,6 +2863,18 @@ async def ui_view_users(
# Using shared metric helper implementations from common_daily_activity
def resolve_user_daily_activity_entity_ids(
*, user_id: str | None, user_api_key_dict: UserAPIKeyAuth
) -> tuple[str, ...] | None | ScopeDenied:
if _user_has_admin_view(user_api_key_dict):
return (user_id,) if user_id is not None else None
caller_user_id: Final = require_caller_user_id_for_non_admin(user_api_key_dict)
if user_id is not None and user_id != caller_user_id:
return ScopeDenied(403, "Non-admin users can only view their own spend data.")
return (caller_user_id,)
async def _resolve_user_email_metadata(
prisma_client: "PrismaClient", records: Sequence[DailySpendRecord]
) -> dict[str, dict]:
@ -2958,20 +2969,13 @@ async def get_user_daily_activity(
)
try:
is_admin: Final = _user_has_admin_view(user_api_key_dict)
if is_admin:
entity_id = user_id # None means global view, otherwise filter by user
else:
caller_user_id: Final = require_caller_user_id_for_non_admin(user_api_key_dict)
if user_id is None:
user_id = caller_user_id
if user_id != caller_user_id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail={"error": "Non-admin users can only view their own spend data."},
)
entity_id = user_id
resolved_entity_ids: Final = resolve_user_daily_activity_entity_ids(
user_id=user_id,
user_api_key_dict=user_api_key_dict,
)
if isinstance(resolved_entity_ids, ScopeDenied):
raise_public(resolved_entity_ids)
entity_id: Final[str | None] = resolved_entity_ids[0] if resolved_entity_ids is not None else None
return await get_daily_activity(
prisma_client=prisma_client,
@ -2998,112 +3002,3 @@ async def get_user_daily_activity(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail={"error": f"Failed to fetch analytics: {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: str | None = fastapi.Query(
default=None,
description="Start date in YYYY-MM-DD format",
),
end_date: str | None = fastapi.Query(
default=None,
description="End date in YYYY-MM-DD format",
),
model: str | None = fastapi.Query(
default=None,
description="Filter by specific model",
),
api_key: str | None = fastapi.Query(
default=None,
description="Filter by specific API key",
),
user_id: str | None = fastapi.Query(
default=None,
description="Filter by specific user ID. Admins can filter by any user or omit for global view. Non-admins must provide their own user_id.",
),
timezone: int | None = fastapi.Query(
default=None,
description="Timezone offset in minutes from UTC (e.g., 480 for PST). "
"Matches JavaScript's Date.getTimezoneOffset() convention.",
),
include_current_utc_day: bool = fastapi.Query(
default=False,
description="When the range ends on the caller's current local day, extend it to "
"today's UTC bucket so spend written after the caller's local midnight (in UTC "
"terms) is included. Requires the timezone parameter. Historical ranges are "
"never extended.",
),
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.
Reads daily spend records that only ever accumulate and are never affected by budget
resets. Their total can legitimately exceed the `spend` field returned by
`/v2/user/info`, which is a running budget counter that every budget reset sets back
to zero (or to the overage above `max_budget` when `budget_rollover` is enabled).
"""
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:
is_admin: Final = _user_has_admin_view(user_api_key_dict)
if is_admin:
entity_id = user_id # None means global view, otherwise filter by user
else:
caller_user_id: Final = require_caller_user_id_for_non_admin(user_api_key_dict)
if user_id is None:
user_id = caller_user_id
if user_id != caller_user_id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail={"error": "Non-admin users can only view their own spend data."},
)
entity_id = user_id
repository: Final = daily_activity_repository(prisma_client)
scope: Final = daily_activity_scope(
"litellm_dailyuserspend",
"user_id",
entity_id,
None,
api_key,
start_date,
end_date,
model,
timezone,
include_current_utc_day,
)
return await get_daily_activity_aggregated(
repository,
scope,
)
except HTTPException:
raise
except Exception as e:
verbose_proxy_logger.exception("/user/daily/activity/aggregated: Exception occured - %s", e)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail={"error": f"Failed to fetch analytics: {e}"},
)

89
litellm/proxy/management_endpoints/organization_endpoints.py
View file

@ -14,10 +14,12 @@ Endpoints for /organization operations
#### ORGANIZATION MANAGEMENT ####
from collections.abc import Mapping, Sequence
from types import MappingProxyType
from typing import (
TYPE_CHECKING,
Annotated,
Final,
NamedTuple,
Protocol,
cast, # noqa: TID251 # prisma types Json columns as fields.Json but reads back plain python values
overload,
@ -62,6 +64,7 @@ from litellm.proxy.management_helpers.utils import (
)
from litellm.proxy.utils import PrismaClient, ProxyLogging
from litellm.repositories.budget_repository import BudgetRepository
from litellm.repositories.chunked_in import find_many_in
from litellm.repositories.object_permission_repository import ObjectPermissionRepository
from litellm.repositories.organization_repository import OrganizationRepository
from litellm.repositories.table_repositories import OrganizationMembershipRepository
@ -583,43 +586,23 @@ async def get_organization_daily_activity(
detail={"error": CommonProxyErrors.db_not_connected_error.value},
)
# Parse comma-separated ids
org_ids_list = organization_ids.split(",") if organization_ids else None
org_ids: Final = tuple(organization_ids.split(",")) if organization_ids else None
exclude_org_ids_list: list[str] | None = None
if exclude_organization_ids:
exclude_org_ids_list = exclude_organization_ids.split(",") if exclude_organization_ids else None
# Restrict non-proxy-admins to only organizations where they are org_admin
if not _user_has_admin_view(user_api_key_dict):
memberships: Final = await _table(OrganizationMembershipRepository(prisma_client)).find_many(
where={"user_id": user_api_key_dict.user_id}
)
admin_org_ids = [m.organization_id for m in memberships if m.user_role == LitellmUserRoles.ORG_ADMIN.value]
if org_ids_list is None:
# Default to orgs where user is org_admin
org_ids_list = admin_org_ids
else:
# Ensure user is org_admin for all requested orgs
for org_id in org_ids_list:
if org_id not in admin_org_ids:
raise HTTPException(
status_code=403,
detail={"error": f"User is not org_admin for Organization= {org_id}."},
)
org_scope: Final = await resolve_organization_daily_activity_scope(
organization_ids=org_ids,
prisma_client=prisma_client,
user_api_key_dict=user_api_key_dict,
)
# Fetch organization aliases for metadata
where_condition: Final = _STR_OBJECT_DICT_ADAPTER.validate_python({})
if org_ids_list is not None:
where_condition["organization_id"] = {"in": list(org_ids_list)}
org_aliases: Final = await _table(OrganizationRepository(prisma_client)).find_many(where=where_condition)
# Query daily activity for organizations
return await get_daily_activity(
prisma_client=prisma_client,
table_name="litellm_dailyorganizationspend",
entity_id_field="organization_id",
entity_id=org_ids_list,
entity_metadata_field={o.organization_id: {"organization_alias": o.organization_alias} for o in org_aliases},
entity_id=None if org_scope.organization_ids is None else list(org_scope.organization_ids),
entity_metadata_field=org_scope.organization_metadata,
exclude_entity_ids=exclude_org_ids_list,
start_date=start_date,
end_date=end_date,
@ -630,6 +613,56 @@ async def get_organization_daily_activity(
)
class _OrganizationDailyActivityScope(NamedTuple):
organization_ids: tuple[str, ...] | None
organization_metadata: Mapping[str, dict[str, object]]
async def resolve_organization_daily_activity_scope(
*,
organization_ids: tuple[str, ...] | None,
prisma_client: PrismaClient,
user_api_key_dict: UserAPIKeyAuth,
) -> _OrganizationDailyActivityScope:
is_admin: Final = _user_has_admin_view(user_api_key_dict)
memberships: Final = (
await _table(OrganizationMembershipRepository(prisma_client)).find_many(
where={"user_id": user_api_key_dict.user_id}
)
if not is_admin
else ()
)
admin_organization_ids: Final = tuple(
membership.organization_id
for membership in memberships
if membership.user_role == LitellmUserRoles.ORG_ADMIN.value
)
if not is_admin and organization_ids is not None:
for organization_id in organization_ids:
if organization_id not in admin_organization_ids:
raise HTTPException(
status_code=403,
detail={"error": f"User is not org_admin for Organization= {organization_id}."},
)
resolved_organization_ids: Final[tuple[str, ...] | None] = (
organization_ids if is_admin or organization_ids is not None else admin_organization_ids
)
organization_table: Final = _table(OrganizationRepository(prisma_client))
organization_rows: Final = (
await find_many_in(organization_table, "organization_id", resolved_organization_ids)
if resolved_organization_ids is not None
else await organization_table.find_many(where={})
)
metadata: Final = MappingProxyType(
{
organization.organization_id: {"organization_alias": organization.organization_alias}
for organization in organization_rows
}
)
return _OrganizationDailyActivityScope(resolved_organization_ids, metadata)
async def _set_object_permission(
data: NewOrganizationRequest,
prisma_client: PrismaClient | None,

4
litellm/proxy/management_endpoints/tag_management_endpoints.py
View file

@ -190,7 +190,7 @@ async def _get_tag_list_scope(
return {"api_key": {"in": scoped_api_keys}}
async def _get_tag_daily_activity_api_key_filter(
async def get_tag_daily_activity_api_key_filter(
prisma_client: "PrismaClient",
user_api_key_dict: UserAPIKeyAuth,
requested_api_key: str | None,
@ -757,7 +757,7 @@ async def get_tag_daily_activity(
# Convert comma-separated tags string to list if provided
tag_list: Final = tags.split(",") if tags else None
scoped_api_key_filter: Final = await _get_tag_daily_activity_api_key_filter(
scoped_api_key_filter: Final = await get_tag_daily_activity_api_key_filter(
prisma_client=prisma_client,
user_api_key_dict=user_api_key_dict,
requested_api_key=api_key,

102
litellm/proxy/management_endpoints/team_endpoints.py
View file

@ -124,11 +124,6 @@ from litellm.proxy.hooks.model_max_budget_limiter import (
)
from litellm.proxy.management.teams.access import TEAM_OR_ORG_ADMIN, TeamRole, is_team_admin, team_access_denied
from litellm.proxy.management.teams.dependencies import get_team_access
from litellm.proxy.management_endpoints.common_daily_activity import (
daily_activity_repository,
daily_activity_scope,
get_daily_activity_aggregated,
)
from litellm.proxy.management_endpoints.common_utils import (
_check_disable_global_guardrails_caller_permission,
_check_passthrough_routes_caller_permission,
@ -6523,7 +6518,7 @@ class _TeamDailyActivityScope(NamedTuple):
api_key_filter: str | list[str] | None # mutable-ok: downstream daily-activity signatures take str | list unions
async def _resolve_team_daily_activity_scope(
async def resolve_team_daily_activity_scope(
*,
team_ids: str | None,
exclude_team_ids: str | None,
@ -6606,11 +6601,14 @@ async def _resolve_team_daily_activity_scope(
user_api_keys = [key.token for key in user_keys if key.token]
# If user has no API keys, return empty result
if not user_api_keys:
user_api_keys = [""] # Use empty string to ensure no matches
user_api_keys = []
# If api_key parameter is provided, use it; otherwise use user_api_keys if set
final_api_key_filter: str | list[str] | None = api_key
if final_api_key_filter is None and user_api_keys is not None:
final_api_key_filter: str | list[str] | None
if user_api_keys is None:
final_api_key_filter = api_key
elif api_key:
final_api_key_filter = api_key if api_key in user_api_keys else []
else:
final_api_key_filter = user_api_keys
return _TeamDailyActivityScope(
@ -6661,7 +6659,7 @@ async def get_team_daily_activity(
if prisma_client is None:
raise _daily_activity_error(status_code=500, message=CommonProxyErrors.db_not_connected_error.value)
scope: Final = await _resolve_team_daily_activity_scope(
scope: Final = await resolve_team_daily_activity_scope(
team_ids=team_ids,
exclude_team_ids=exclude_team_ids,
api_key=api_key,
@ -6690,7 +6688,7 @@ async def get_team_daily_activity(
_MAX_AGGREGATED_RANGE_DAYS: Final = 400
def _aggregated_date_range_error(start_date: str | None, end_date: str | None) -> str | None:
def aggregated_date_range_error(start_date: str | None, end_date: str | None) -> str | None:
"""The aggregated endpoint has no pagination to bound its work, so malformed
dates and ranges wider than the UI ever requests are rejected before querying."""
if start_date is None or end_date is None:
@ -6707,82 +6705,6 @@ def _aggregated_date_range_error(start_date: str | None, end_date: str | None) -
return None
@router.get(
"/team/daily/activity/aggregated",
response_model=SpendAnalyticsPaginatedResponse,
tags=["team management"],
)
async def get_team_daily_activity_aggregated(
user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
team_ids: str | None = None,
start_date: str | None = None,
end_date: str | None = None,
model: str | None = None,
api_key: str | None = None,
exclude_team_ids: str | None = None,
timezone: int | None = None,
):
"""
Aggregated daily activity for teams without pagination, including per-team breakdown.
One SQL GROUPING SETS pass returns every day in the range regardless of row
volume, so callers never reassemble pages. Same response shape as the
paginated endpoint with page metadata pinned to a single page.
Args:
team_ids (Optional[str]): Comma-separated list of team IDs to filter by. If not provided, returns data for all teams.
start_date (Optional[str]): Start date for the activity period (YYYY-MM-DD).
end_date (Optional[str]): End date for the activity period (YYYY-MM-DD).
model (Optional[str]): Filter by model name.
api_key (Optional[str]): Filter by API key.
exclude_team_ids (Optional[str]): Comma-separated list of team IDs to exclude.
timezone (Optional[int]): Timezone offset in minutes from UTC, matching JavaScript's Date.getTimezoneOffset() convention.
Returns:
SpendAnalyticsPaginatedResponse: Response containing all daily activity data for the range.
"""
from litellm.proxy.proxy_server import (
prisma_client,
proxy_logging_obj,
user_api_key_cache,
)
if prisma_client is None:
raise _daily_activity_error(status_code=500, message=CommonProxyErrors.db_not_connected_error.value)
range_error: Final = _aggregated_date_range_error(start_date, end_date)
if range_error is not None:
raise _daily_activity_error(status_code=400, message=range_error)
scope: Final = await _resolve_team_daily_activity_scope(
team_ids=team_ids,
exclude_team_ids=exclude_team_ids,
api_key=api_key,
user_api_key_dict=user_api_key_dict,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
proxy_logging_obj=proxy_logging_obj,
)
repository: Final = daily_activity_repository(prisma_client)
activity_scope: Final = daily_activity_scope(
"litellm_dailyteamspend",
"team_id",
scope.team_ids,
scope.exclude_team_ids,
scope.api_key_filter,
start_date,
end_date,
model,
timezone,
)
return await get_daily_activity_aggregated(
repository,
activity_scope,
entity_metadata_field=scope.team_alias_metadata,
include_entity_breakdown=True,
)
def _team_user_spend_sql(*, team_count: int, restrict_to_user: bool) -> str:
team_placeholders: Final = ", ".join(f"${i}" for i in range(3, 3 + team_count))
user_clause: Final = f' AND sl."user" = ${3 + team_count}' if restrict_to_user else ""
@ -6850,14 +6772,14 @@ async def get_team_spend_by_user(
if prisma_client is None:
raise _daily_activity_error(status_code=500, message=CommonProxyErrors.db_not_connected_error.value)
range_error: Final = _aggregated_date_range_error(start_date, end_date)
range_error: Final = aggregated_date_range_error(start_date, end_date)
if range_error is not None or start_date is None or end_date is None:
raise _daily_activity_error(status_code=400, message=range_error or "Please provide start_date and end_date")
if not team_ids:
raise _daily_activity_error(status_code=400, message="Please provide team_ids")
scope: Final = await _resolve_team_daily_activity_scope(
scope: Final = await resolve_team_daily_activity_scope(
team_ids=team_ids,
exclude_team_ids=None,
api_key=None,

2
litellm/proxy/proxy_server.py
View file

@ -604,6 +604,7 @@ from litellm.proxy.management_endpoints.cost_tracking_settings import (
from litellm.proxy.management_endpoints.customer_endpoints import (
router as customer_router,
)
from litellm.proxy.management_endpoints.daily_activity_routes import router as daily_activity_router
from litellm.proxy.management_endpoints.fallback_management_endpoints import (
router as fallback_management_router,
)
@ -19951,6 +19952,7 @@ app.include_router(pass_through_router)
app.include_router(health_router)
app.include_router(key_management_router)
app.include_router(internal_user_router)
app.include_router(daily_activity_router)
app.include_router(password_management_router)
app.include_router(session_management_router)
app.include_router(team_router)

45
litellm/types/proxy/management_endpoints/common_daily_activity.py
View file

@ -124,6 +124,51 @@ class SpendAnalyticsPaginatedResponse(BaseModel):
metadata: DailySpendMetadata = Field(default_factory=DailySpendMetadata)
class KeyActivityRow(BaseModel):
api_key: str
metrics: SpendMetrics
metadata: KeyMetadata
class KeySpendMetrics(BaseModel):
spend: float = 0.0
prompt_tokens: int = 0
completion_tokens: int = 0
total_tokens: int = 0
api_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
cache_read_input_tokens: int = 0
cache_creation_input_tokens: int = 0
class KeySpendActivityRow(BaseModel):
api_key: str
metrics: KeySpendMetrics
metadata: KeyMetadata
class DailyActivityKeySearchResponse(BaseModel):
api_keys: list[KeyActivityRow]
class DailyActivityKeyPageResponse(BaseModel):
api_keys: list[KeySpendActivityRow]
total_api_keys: int
offset: int
limit: int
class ModelTopKeysResponse(BaseModel):
model: str
by_model_group: bool
api_keys: list[KeySpendActivityRow]
class CacheLeakageKeysResponse(BaseModel):
api_keys: list[KeySpendActivityRow]
class LiteLLM_DailyUserSpend(BaseModel):
id: str
user_id: str

29
terraform/provider/tools/endpointaudit/coverage_allowlist.txt
View file

@ -12,14 +12,34 @@
# Read-only analytics and spend reporting; observability, not Terraform-managed state
GET /agent/daily/activity
GET /agent/daily/activity/aggregated
GET /agent/daily/activity/aggregated/keys
GET /agent/daily/activity/aggregated/model_top_keys
GET /agent/daily/activity/aggregated/search
GET /agent/daily/activity/export
GET /customer/daily/activity
GET /customer/daily/activity/aggregated
GET /customer/daily/activity/aggregated/keys
GET /customer/daily/activity/aggregated/model_top_keys
GET /customer/daily/activity/aggregated/search
GET /customer/daily/activity/export
GET /guardrails/usage/detail/{guardrail_id}
GET /guardrails/usage/logs
GET /guardrails/usage/overview
GET /key/spend/report
GET /organization/daily/activity
GET /organization/daily/activity/aggregated
GET /organization/daily/activity/aggregated/keys
GET /organization/daily/activity/aggregated/model_top_keys
GET /organization/daily/activity/aggregated/search
GET /organization/daily/activity/export
GET /organization/spend/report
GET /tag/daily/activity
GET /tag/daily/activity/aggregated
GET /tag/daily/activity/aggregated/keys
GET /tag/daily/activity/aggregated/model_top_keys
GET /tag/daily/activity/aggregated/search
GET /tag/daily/activity/export
GET /tag/dau
GET /tag/distinct
GET /tag/mau
@ -28,10 +48,19 @@ GET /tag/user-agent/per-user-analytics
GET /tag/wau
GET /team/daily/activity
GET /team/daily/activity/aggregated
GET /team/daily/activity/aggregated/keys
GET /team/daily/activity/aggregated/model_top_keys
GET /team/daily/activity/aggregated/search
GET /team/daily/activity/export
GET /team/spend/by_user
GET /team/spend/report
GET /user/daily/activity
GET /user/daily/activity/aggregated
GET /user/daily/activity/aggregated/keys
GET /user/daily/activity/aggregated/cache_leakage_keys
GET /user/daily/activity/aggregated/model_top_keys
GET /user/daily/activity/aggregated/search
GET /user/daily/activity/export
GET /user/spend/report
# Admin UI helper endpoints; serve UI forms and caller-scoped views, not desired state

5
tests/code_coverage_tests/unbounded_in_baseline.txt
View file

@ -11,7 +11,6 @@ litellm/proxy/_experimental/mcp_server/oauth2_flow_backfill.py backfill_null_oau
litellm/proxy/_experimental/mcp_server/oauth2_flow_backfill.py backfill_null_oauth2_flows prisma server_id.in `server_ids` 0
litellm/proxy/_experimental/mcp_server/toolset_db.py list_mcp_toolsets prisma toolset_id.in `toolset_ids` 0
litellm/proxy/agent_endpoints/endpoints.py _attach_keys_to_agents prisma agent_id.in `agent_ids` 0
litellm/proxy/agent_endpoints/endpoints.py get_agent_daily_activity prisma agent_id.in `list(agent_ids_list)` 0
litellm/proxy/agent_endpoints/endpoints.py get_agents prisma agent_id.in `agent_ids` 0
litellm/proxy/anthropic_endpoints/claude_code_endpoints/claude_code_skill_access.py SkillVisibility.where prisma name.in `sorted(self.granted)` 0
litellm/proxy/auth/auth_checks.py _fetch_uncached_model_access_group_budgets prisma access_group_name.in `list(uncached_groups)` 0
@ -43,7 +42,6 @@ litellm/proxy/management_endpoints/common_utils.py _team_admin_can_invite_user p
litellm/proxy/management_endpoints/common_utils.py _user_has_admin_privileges prisma team_id.in `user_obj.teams` 0
litellm/proxy/management_endpoints/customer_endpoints.py delete_end_user prisma user_id.in `data.user_ids` 0
litellm/proxy/management_endpoints/customer_endpoints.py delete_end_user prisma user_id.in `data.user_ids` 1
litellm/proxy/management_endpoints/customer_endpoints.py get_customer_daily_activity prisma user_id.in `list(end_user_ids_list)` 0
litellm/proxy/management_endpoints/internal_user_endpoints.py _check_user_info_v2_access prisma team_id.in `caller_user.teams` 0
litellm/proxy/management_endpoints/internal_user_endpoints.py _resolve_user_email_metadata prisma user_id.in `list(user_ids)` 0
litellm/proxy/management_endpoints/internal_user_endpoints.py delete_user prisma created_by.in `data.user_ids` 0
@ -73,7 +71,6 @@ litellm/proxy/management_endpoints/mcp_management_endpoints.py fetch_all_mcp_ser
litellm/proxy/management_endpoints/model_access_group_management_endpoints.py update_deployments_with_access_group prisma model_name.in `model_names` 0
litellm/proxy/management_endpoints/model_management_endpoints.py delete_team_models prisma model_id.in `model_ids` 0
litellm/proxy/management_endpoints/organization_endpoints.py deprecated_info_organization prisma organization_id.in `data.organizations` 0
litellm/proxy/management_endpoints/organization_endpoints.py get_organization_daily_activity prisma organization_id.in `list(org_ids_list)` 0
litellm/proxy/management_endpoints/organization_endpoints.py list_organization prisma organization_id.in `membership_org_ids` 0
litellm/proxy/management_endpoints/router_weights.py validate_router_settings_weights prisma model_id.in `list(deployment_ids)` 0
litellm/proxy/management_endpoints/session_endpoints.py revoke_ui_session_keys prisma token.in `revoked_tokens` 0
@ -91,7 +88,7 @@ litellm/proxy/management_endpoints/team_endpoints.py _build_team_list_where_cond
litellm/proxy/management_endpoints/team_endpoints.py _get_keys_count_by_team prisma team_id.in `page_team_ids` 0
litellm/proxy/management_endpoints/team_endpoints.py _hydrate_member_user_details prisma user_id.in `sorted(user_ids)` 0
litellm/proxy/management_endpoints/team_endpoints.py _resolve_existing_member_user_ids prisma user_id.in `sorted(requested_user_ids)` 0
litellm/proxy/management_endpoints/team_endpoints.py _resolve_team_daily_activity_scope prisma team_id.in `list(team_ids_list)` 0
litellm/proxy/management_endpoints/team_endpoints.py resolve_team_daily_activity_scope prisma team_id.in `list(team_ids_list)` 0
litellm/proxy/management_endpoints/team_endpoints.py _sweep_deleted_team_references prisma team_id.in `tuple(team_ids)` 0
litellm/proxy/management_endpoints/team_endpoints.py _sweep_deleted_team_references_tx prisma team_id.in `tuple(team_ids)` 0
litellm/proxy/management_endpoints/team_endpoints.py _team_member_delete prisma user_id.in `sorted(addressed_user_ids)` 0

1169
tests/integration/spend/golden/daily_activity_team_aggregated.json Normal file
View file

File diff suppressed because it is too large Load diff

1197
tests/integration/spend/golden/daily_activity_team_paginated.json Normal file
View file

File diff suppressed because it is too large Load diff

1002
tests/integration/spend/golden/daily_activity_user_aggregated.json Normal file
View file

File diff suppressed because it is too large Load diff

1198
tests/integration/spend/golden/daily_activity_user_paginated.json Normal file
View file

File diff suppressed because it is too large Load diff

514
tests/integration/spend/test_daily_activity_routes.py Normal file
View file

@ -0,0 +1,514 @@
import csv
import hashlib
import io
import uuid
from datetime import datetime, timedelta, timezone
from itertools import chain
from pathlib import Path
from typing import Final
import httpx
import pytest
from fastapi import FastAPI
from integration._support.client import JSON_OBJECT, Gateway, eventually, object_value, string_value
from integration._support.database import read_rows
from integration._support.process import owned_proxy
from integration._support.upstream import JsonResponse, delete_scenario, register_scenario
from integration.spend.test_daily_activity_repository import _daily_activity_database, _PrismaDatabase, _repository
from litellm import constants
from litellm.proxy import proxy_server
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.management_endpoints.daily_activity_routes import (
get_daily_activity_prisma_client,
get_daily_activity_repository,
)
from litellm.proxy.management_endpoints.daily_activity_routes import (
router as daily_activity_router,
)
from litellm.proxy.management_endpoints.internal_user_endpoints import router as internal_user_router
from litellm.proxy.management_endpoints.team_endpoints import router as team_router
from litellm.types.proxy.management_endpoints.common_daily_activity import DailyActivityKeyPageResponse
def _delete_organization(gateway: Gateway, organization_id: str) -> None:
response: Final = gateway.request("DELETE", "/organization/delete", {"organization_ids": [organization_id]})
assert response.status_code == 200, response.text
def _delete_tag(gateway: Gateway, tag: str) -> None:
response: Final = gateway.request("POST", "/tag/delete", {"name": tag})
assert response.status_code == 200, response.text
def _delete_end_user(gateway: Gateway, end_user_id: str) -> None:
response: Final = gateway.request("POST", "/end_user/delete", {"user_ids": [end_user_id]})
assert response.status_code == 200, response.text
def _delete_agent(gateway: Gateway, agent_id: str) -> None:
response: Final = gateway.request("DELETE", f"/v1/agents/{agent_id}")
assert response.status_code == 200, response.text
def _daily_activity_request(
gateway: Gateway,
*,
model: str,
key: str,
end_user_id: str,
tag: str,
request_number: int,
) -> None:
response: Final = gateway.request(
"POST",
"/v1/chat/completions",
JSON_OBJECT.validate_python(
{
"model": model,
"messages": [{"role": "user", "content": f"daily activity {request_number}"}],
"metadata": {"tags": [tag]},
"user": end_user_id,
}
),
key=key,
)
assert response.status_code == 200, response.text
def _aggregate_result_api_keys(result: object) -> tuple[str, ...]:
result_body: Final = object_value(result)
breakdown: Final = object_value(result_body["breakdown"])
api_keys: Final = object_value(breakdown["api_keys"])
return tuple(api_keys)
def _aggregate_top_keys(results: object) -> frozenset[str]:
assert isinstance(results, list)
api_keys_by_result: Final = tuple(_aggregate_result_api_keys(result) for result in results)
return frozenset(chain.from_iterable(api_keys_by_result))
def _assert_entity_activity_routes(
gateway: Gateway,
*,
prefix: str,
entity_param: str,
entity_id: str,
table: str,
entity_column: str,
date_params: dict[str, str],
target_digest: str,
model: str,
) -> None:
params: Final = {**date_params, entity_param: entity_id}
persisted_rows: Final = eventually(
lambda: read_rows(
f'SELECT api_key FROM "{table}" WHERE "{entity_column}"=%s AND date BETWEEN %s AND %s',
(entity_id, date_params["start_date"], date_params["end_date"]),
),
lambda rows: len(rows) == 6,
seconds=70,
)
aggregated: Final = gateway.request(
"GET",
f"{prefix}/daily/activity/aggregated",
params={**params, "api_key_limit": "3"},
)
assert aggregated.status_code == 200, aggregated.text
aggregate_body: Final = object_value(aggregated.json())
metadata: Final = object_value(aggregate_body["metadata"])
total_api_keys: Final = metadata["total_api_keys"]
api_key_limit: Final = metadata["api_key_limit"]
assert isinstance(total_api_keys, int) and total_api_keys == 6, aggregated.text
assert isinstance(api_key_limit, int) and api_key_limit == 3, aggregated.text
assert total_api_keys > api_key_limit, aggregated.text
assert metadata["total_api_requests"] == 8, aggregated.text
top_api_keys: Final = _aggregate_top_keys(aggregate_body["results"])
assert target_digest not in top_api_keys, aggregated.text
ranked_rows: Final = read_rows(
f'SELECT api_key FROM "{table}" WHERE "{entity_column}"=%s AND date BETWEEN %s AND %s '
"AND api_key <> %s GROUP BY api_key ORDER BY SUM(spend::numeric) DESC, api_key",
(
entity_id,
date_params["start_date"],
date_params["end_date"],
constants.PTU_SENTINEL_API_KEY,
),
)
ranked_keys: Final = tuple(string_value(row["api_key"]) for row in ranked_rows)
page_responses: Final = tuple(
gateway.request(
"GET",
f"{prefix}/daily/activity/aggregated/keys",
params={**params, "offset": str(offset), "limit": "2"},
)
for offset in range(0, len(ranked_keys), 2)
)
assert all(response.status_code == 200 for response in page_responses), tuple(
response.text for response in page_responses
)
page_bodies: Final = tuple(
DailyActivityKeyPageResponse.model_validate_json(response.content) for response in page_responses
)
page_api_keys: Final = tuple(tuple(row.api_key for row in body.api_keys) for body in page_bodies)
paged_keys: Final = tuple(chain.from_iterable(page_api_keys))
assert tuple(body.total_api_keys for body in page_bodies) == (6,) * len(page_bodies)
assert paged_keys == ranked_keys
assert len(paged_keys) == len(frozenset(paged_keys))
assert frozenset(paged_keys[:3]) == top_api_keys, aggregated.text
key_details: Final = gateway.request(
"GET",
f"{prefix}/daily/activity/aggregated",
params={**params, "api_key": target_digest},
)
assert key_details.status_code == 200, key_details.text
key_details_body: Final = JSON_OBJECT.validate_json(key_details.content)
assert object_value(key_details_body["metadata"])["total_api_keys"] == 1, key_details.text
assert _aggregate_top_keys(key_details_body["results"]) == frozenset((target_digest,)), key_details.text
searched: Final = gateway.request(
"GET",
f"{prefix}/daily/activity/aggregated/search",
params={**params, "search": target_digest},
)
assert searched.status_code == 200, searched.text
search_body: Final = object_value(searched.json())
search_rows: Final = search_body["api_keys"]
assert isinstance(search_rows, list) and len(search_rows) == 1, searched.text
assert object_value(search_rows[0])["api_key"] == target_digest, searched.text
top_keys: Final = gateway.request(
"GET",
f"{prefix}/daily/activity/aggregated/model_top_keys",
params={**params, "model_group": model},
)
assert top_keys.status_code == 200, top_keys.text
top_body: Final = object_value(top_keys.json())
top_rows: Final = top_body["api_keys"]
assert isinstance(top_rows, list) and len(top_rows) == 5, top_keys.text
top_spends: Final = tuple(object_value(object_value(row)["metrics"])["spend"] for row in top_rows[:2])
assert top_spends == (
pytest.approx(0.12),
pytest.approx(0.12),
), top_keys.text
exported: Final = gateway.request(
"GET",
f"{prefix}/daily/activity/export",
params={**params, "export_type": "daily_with_keys"},
)
assert exported.status_code == 200, exported.text
export_rows: Final = tuple(csv.reader(io.StringIO(exported.text)))
assert len(export_rows) == len(persisted_rows) + 1, exported.text
@pytest.mark.timeout(90)
def test_daily_activity_routes_cover_all_entities_and_bounded_key_search(gateway: Gateway, tmp_path: Path) -> None:
with owned_proxy(gateway, tmp_path, {}) as proxy:
_assert_daily_activity_routes(proxy)
def _assert_daily_activity_routes(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
model: Final = scenario.model(input_cost_per_token=0.001, output_cost_per_token=0.002)
target_scenario_id: Final = f"usage-cache-{uuid.uuid4().hex}"
target_response: Final = JsonResponse(
content_type="application/json",
body=JSON_OBJECT.validate_python(
{
"id": "$UNIQUE_ID",
"object": "chat.completion",
"created": 1_700_000_000,
"model": "gpt-4o-mini",
"choices": [
{
"index": 0,
"message": {"role": "assistant", "content": "cached response"},
"finish_reason": "stop",
}
],
"usage": {
"prompt_tokens": 40,
"completion_tokens": 20,
"total_tokens": 60,
"prompt_tokens_details": {"cached_tokens": 20},
},
}
),
)
target_upstream: Final = register_scenario(target_scenario_id, target_response)
scenario.cleanups.callback(delete_scenario, target_upstream)
cache_model: Final = scenario.model(
api_base=target_upstream.api_base(),
api_key=target_scenario_id,
input_cost_per_token=0.0,
output_cost_per_token=0.0,
)
organization: Final = gateway.post(
"/organization/new",
{"organization_alias": f"integration-{uuid.uuid4().hex}"},
)
organization_id: Final = string_value(organization["organization_id"])
scenario.cleanups.callback(_delete_organization, gateway, organization_id)
team_id: Final = scenario.team(organization_id=organization_id)
user_id: Final = scenario.user()
tag: Final = f"integration-{uuid.uuid4().hex}"
gateway.post("/tag/new", {"name": tag})
scenario.cleanups.callback(_delete_tag, gateway, tag)
end_user_id: Final = f"integration-{uuid.uuid4().hex}"
gateway.post("/end_user/new", {"user_id": end_user_id})
scenario.cleanups.callback(_delete_end_user, gateway, end_user_id)
agent_response: Final = gateway.request(
"POST",
"/v1/agents",
{
"agent_name": f"integration-{uuid.uuid4().hex}",
"agent_card_params": {
"protocolVersion": "0.3",
"name": "integration",
"description": "integration agent",
"url": "http://127.0.0.1:1/agent",
"version": "1",
"capabilities": {},
"defaultInputModes": ["text"],
"defaultOutputModes": ["text"],
"skills": [],
},
},
)
assert agent_response.status_code == 200, agent_response.text
agent_id: Final = string_value(object_value(agent_response.json())["agent_id"])
scenario.cleanups.callback(_delete_agent, gateway, agent_id)
keys: Final = tuple(
scenario.key(
models=[model, cache_model],
team_id=team_id,
user_id=user_id,
organization_id=organization_id,
agent_id=agent_id,
)
for _ in range(5)
)
target_key: Final = scenario.key(
models=[model, cache_model],
team_id=team_id,
user_id=user_id,
organization_id=organization_id,
agent_id=agent_id,
)
for request_number, key in enumerate(keys):
_daily_activity_request(
gateway,
model=model,
key=key,
end_user_id=end_user_id,
tag=tag,
request_number=request_number,
)
for request_number, key in enumerate(keys[:2]):
_daily_activity_request(
gateway,
model=model,
key=key,
end_user_id=end_user_id,
tag=tag,
request_number=100 + request_number,
)
target_request: Final = gateway.request(
"POST",
"/v1/chat/completions",
JSON_OBJECT.validate_python(
{
"model": cache_model,
"messages": [{"role": "user", "content": "cached activity response"}],
"metadata": {"tags": [tag]},
"user": end_user_id,
}
),
key=target_key,
)
assert target_request.status_code == 200, target_request.text
today: Final = datetime.now(timezone.utc).date()
start_date: Final = (today - timedelta(days=1)).isoformat()
end_date: Final = (today + timedelta(days=1)).isoformat()
date_params: Final = {"start_date": start_date, "end_date": end_date, "timezone": "0"}
route_cases: Final = (
("/user", "user_id", user_id, "LiteLLM_DailyUserSpend", "user_id"),
("/team", "team_ids", team_id, "LiteLLM_DailyTeamSpend", "team_id"),
("/tag", "tags", tag, "LiteLLM_DailyTagSpend", "tag"),
(
"/organization",
"organization_ids",
organization_id,
"LiteLLM_DailyOrganizationSpend",
"organization_id",
),
("/customer", "end_user_ids", end_user_id, "LiteLLM_DailyEndUserSpend", "end_user_id"),
("/agent", "agent_ids", agent_id, "LiteLLM_DailyAgentSpend", "agent_id"),
)
target_digest: Final = hashlib.sha256(target_key.encode()).hexdigest()
for prefix, entity_param, entity_id, table, entity_column in route_cases:
_assert_entity_activity_routes(
gateway,
prefix=prefix,
entity_param=entity_param,
entity_id=entity_id,
table=table,
entity_column=entity_column,
date_params=date_params,
target_digest=target_digest,
model=model,
)
user_cache_keys: Final = gateway.request(
"GET",
"/user/daily/activity/aggregated/cache_leakage_keys",
params={**date_params, "user_id": user_id},
)
assert user_cache_keys.status_code == 200, user_cache_keys.text
cache_rows: Final = object_value(user_cache_keys.json())["api_keys"]
assert isinstance(cache_rows, list) and cache_rows, user_cache_keys.text
cache_api_keys: Final = tuple(string_value(object_value(row)["api_key"]) for row in cache_rows)
assert target_digest in cache_api_keys, user_cache_keys.text
async def _assert_route_matches_golden(client: httpx.AsyncClient, route: str, golden_name: str) -> None:
response: Final = await client.get(
route,
params={"start_date": "2026-06-01", "end_date": "2026-06-01"},
)
assert response.status_code == 200, response.text
golden: Final = (Path(__file__).parent / "golden" / golden_name).read_text()
expected: Final = JSON_OBJECT.validate_json(golden)
actual: Final = object_value(response.json())
assert actual == expected, route
@pytest.mark.asyncio
async def test_existing_activity_routes_match_base_branch_goldens(monkeypatch: pytest.MonkeyPatch) -> None:
async with _daily_activity_database() as database:
repository: Final = _repository(database)
app: Final = FastAPI()
app.include_router(internal_user_router)
app.include_router(team_router)
app.include_router(daily_activity_router)
monkeypatch.setattr(proxy_server, "prisma_client", _PrismaDatabase(database))
app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth(
user_id="integration-admin", user_role=LitellmUserRoles.PROXY_ADMIN
)
app.dependency_overrides[get_daily_activity_prisma_client] = lambda: _PrismaDatabase(database)
app.dependency_overrides[get_daily_activity_repository] = lambda: repository
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://testserver") as client:
route_goldens: Final = (
("/user/daily/activity", "daily_activity_user_paginated.json"),
("/user/daily/activity/aggregated", "daily_activity_user_aggregated.json"),
("/team/daily/activity", "daily_activity_team_paginated.json"),
("/team/daily/activity/aggregated", "daily_activity_team_aggregated.json"),
)
for route, golden_name in route_goldens:
await _assert_route_matches_golden(client, route, golden_name)
@pytest.mark.asyncio
async def test_user_key_pages_and_details_respect_caller_scope() -> None:
async with _daily_activity_database() as database:
await database.query_raw(
'INSERT INTO "LiteLLM_UserTable" (user_id, user_email, models) VALUES ($1, $2, $3)',
"user-2",
"other@example.test",
[],
)
await database.query_raw(
"""
INSERT INTO "LiteLLM_VerificationToken"
(token, key_alias, team_id, user_id, metadata, models)
VALUES ($1, $2, $3, $4, $5::jsonb, $6)
""",
"key-other-user",
"Other user key",
None,
"user-2",
"{}",
[],
)
await database.query_raw(
"""
INSERT INTO "LiteLLM_DailyUserSpend"
(id, user_id, date, api_key, model, model_group, custom_llm_provider,
mcp_namespaced_tool_name, endpoint, prompt_tokens, completion_tokens,
cache_read_input_tokens, cache_creation_input_tokens, spend, api_requests,
successful_requests, failed_requests, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18::timestamp)
""",
"other-user-row",
"user-2",
"2026-06-01",
"key-other-user",
"model",
"",
"provider-a",
None,
"/v1/chat/completions",
1,
1,
0,
0,
50.0,
1,
1,
0,
"2026-06-01 12:00:00",
)
repository: Final = _repository(database)
app: Final = FastAPI()
app.include_router(daily_activity_router)
app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth(
user_id="user-1",
user_role=LitellmUserRoles.INTERNAL_USER,
)
app.dependency_overrides[get_daily_activity_prisma_client] = lambda: _PrismaDatabase(database)
app.dependency_overrides[get_daily_activity_repository] = lambda: repository
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=app),
base_url="http://testserver",
) as client:
params: Final = {"start_date": "2026-06-01", "end_date": "2026-06-01"}
page: Final = await client.get(
"/user/daily/activity/aggregated/keys",
params={**params, "user_id": "user-1", "limit": 100},
)
assert page.status_code == 200, page.text
page_body: Final = DailyActivityKeyPageResponse.model_validate_json(page.content)
page_keys: Final = frozenset(row.api_key for row in page_body.api_keys)
assert page_body.total_api_keys == len(page_keys) == 5
assert "key-other-user" not in page_keys
denied: Final = await client.get(
"/user/daily/activity/aggregated/keys",
params={**params, "user_id": "user-2"},
)
assert denied.status_code == 403, denied.text
own_details: Final = await client.get(
"/user/daily/activity/aggregated",
params={**params, "user_id": "user-1", "api_key": "key-a"},
)
assert own_details.status_code == 200, own_details.text
own_body: Final = JSON_OBJECT.validate_json(own_details.content)
assert object_value(own_body["metadata"])["total_api_keys"] == 1
assert _aggregate_top_keys(own_body["results"]) == frozenset(("key-a",))
other_details: Final = await client.get(
"/user/daily/activity/aggregated",
params={**params, "user_id": "user-1", "api_key": "key-other-user"},
)
assert other_details.status_code == 200, other_details.text
other_body: Final = JSON_OBJECT.validate_json(other_details.content)
assert object_value(other_body["metadata"])["total_api_keys"] == 0
assert _aggregate_top_keys(other_body["results"]) == frozenset()

89
tests/unit/proxy/auth/test_route_checks.py
View file

@ -16,6 +16,93 @@ from litellm.proxy._types import (
from litellm.proxy.auth.auth_checks_organization import _user_is_org_admin
from litellm.proxy.auth.route_checks import RouteChecks
DAILY_ACTIVITY_ROUTE_PAIRS: Final[tuple[tuple[str, str], ...]] = (
("/user/daily/activity", "/user/daily/activity/aggregated"),
("/user/daily/activity", "/user/daily/activity/aggregated/keys"),
("/user/daily/activity", "/user/daily/activity/aggregated/search"),
("/user/daily/activity", "/user/daily/activity/aggregated/model_top_keys"),
("/user/daily/activity", "/user/daily/activity/export"),
("/user/daily/activity", "/user/daily/activity/aggregated/cache_leakage_keys"),
("/team/daily/activity", "/team/daily/activity/aggregated"),
("/team/daily/activity", "/team/daily/activity/aggregated/keys"),
("/team/daily/activity", "/team/daily/activity/aggregated/search"),
("/team/daily/activity", "/team/daily/activity/aggregated/model_top_keys"),
("/team/daily/activity", "/team/daily/activity/export"),
("/tag/daily/activity", "/tag/daily/activity/aggregated"),
("/tag/daily/activity", "/tag/daily/activity/aggregated/keys"),
("/tag/daily/activity", "/tag/daily/activity/aggregated/search"),
("/tag/daily/activity", "/tag/daily/activity/aggregated/model_top_keys"),
("/tag/daily/activity", "/tag/daily/activity/export"),
("/organization/daily/activity", "/organization/daily/activity/aggregated"),
("/organization/daily/activity", "/organization/daily/activity/aggregated/keys"),
("/organization/daily/activity", "/organization/daily/activity/aggregated/search"),
("/organization/daily/activity", "/organization/daily/activity/aggregated/model_top_keys"),
("/organization/daily/activity", "/organization/daily/activity/export"),
("/customer/daily/activity", "/customer/daily/activity/aggregated"),
("/customer/daily/activity", "/customer/daily/activity/aggregated/keys"),
("/customer/daily/activity", "/customer/daily/activity/aggregated/search"),
("/customer/daily/activity", "/customer/daily/activity/aggregated/model_top_keys"),
("/customer/daily/activity", "/customer/daily/activity/export"),
("/customer/daily/activity", "/end_user/daily/activity/aggregated"),
("/customer/daily/activity", "/end_user/daily/activity/aggregated/keys"),
("/customer/daily/activity", "/end_user/daily/activity/aggregated/search"),
("/customer/daily/activity", "/end_user/daily/activity/aggregated/model_top_keys"),
("/customer/daily/activity", "/end_user/daily/activity/export"),
("/agent/daily/activity", "/agent/daily/activity/aggregated"),
("/agent/daily/activity", "/agent/daily/activity/aggregated/keys"),
("/agent/daily/activity", "/agent/daily/activity/aggregated/search"),
("/agent/daily/activity", "/agent/daily/activity/aggregated/model_top_keys"),
("/agent/daily/activity", "/agent/daily/activity/export"),
)
DAILY_ACTIVITY_ROLES: Final[tuple[LitellmUserRoles, ...]] = (
LitellmUserRoles.PROXY_ADMIN,
LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY,
LitellmUserRoles.INTERNAL_USER,
LitellmUserRoles.INTERNAL_USER_VIEW_ONLY,
LitellmUserRoles.ORG_ADMIN,
LitellmUserRoles.TEAM,
LitellmUserRoles.CUSTOMER,
)
def _daily_activity_route_outcome(route: str, user_role: LitellmUserRoles) -> str:
if user_role == LitellmUserRoles.PROXY_ADMIN:
return "allowed"
user_obj = LiteLLM_UserTable(
user_id="test_user",
user_email="test@example.com",
user_role=user_role.value,
)
valid_token = UserAPIKeyAuth(user_id="test_user", user_role=user_role)
request = MagicMock(spec=Request)
request.method = "GET"
request.query_params = {}
try:
RouteChecks.non_proxy_admin_allowed_routes_check(
user_obj=user_obj,
_user_role=user_role.value,
route=route,
request=request,
valid_token=valid_token,
request_data={},
)
except HTTPException as exc:
return f"denied:{exc.status_code}"
except Exception as exc:
return f"denied:{type(exc).__name__}"
return "allowed"
@pytest.mark.parametrize(("existing_path", "new_path"), DAILY_ACTIVITY_ROUTE_PAIRS)
@pytest.mark.parametrize("user_role", DAILY_ACTIVITY_ROLES)
def test_daily_activity_routes_preserve_route_access_outcomes(
existing_path: str, new_path: str, user_role: LitellmUserRoles
) -> None:
assert _daily_activity_route_outcome(new_path, user_role) == _daily_activity_route_outcome(
existing_path, user_role
)
def test_non_admin_config_update_route_rejected():
"""Test that non-admin users are rejected when trying to call /config/update"""
@ -2219,7 +2306,7 @@ def test_internal_user_can_access_logs_drawer_detail_route(user_role):
request_data={},
)
except Exception as e:
pytest.fail(f"{user_role.value} should be able to access {route}. Got error: {str(e)}")
pytest.fail(f"{user_role.value} should be able to access {route}. Got error: {e!s}")
@pytest.mark.parametrize(

56
tests/unit/proxy/management_endpoints/test_activity_tenant_scoping.py
View file

@ -387,3 +387,59 @@ async def test_agent_activity_non_admin_no_access_returns_empty_page():
assert result.results == []
fake_get_daily.assert_not_awaited()
@pytest.mark.asyncio
@pytest.mark.parametrize(
("owned_tokens", "requested_api_key"),
[
([], None),
(["alice-key-1"], "bob-key-1"),
],
)
async def test_team_activity_member_without_matching_keys_queries_nothing(
owned_tokens: list[str], requested_api_key: str | None
) -> None:
"""A member without full team view whose key list is empty, or who asks for
a key they do not own, must reach the repository with an empty key filter,
never with no filter at all."""
from litellm.proxy.management_endpoints import common_daily_activity, team_endpoints
from litellm.repositories.daily_activity_sql import build_where_clause
from litellm.types.repositories.daily_activity import DailyRowsPage
user = UserAPIKeyAuth(user_id="alice", user_role=LitellmUserRoles.INTERNAL_USER.value)
prisma = MagicMock()
prisma.db.litellm_teamtable.find_many = AsyncMock(return_value=[_make_team("team-B", admin_user_ids=["bob"])])
prisma.db.litellm_verificationtoken.find_many = AsyncMock(
return_value=[MagicMock(token=token) for token in owned_tokens]
)
user_info = MagicMock()
user_info.teams = ["team-B"]
repository = MagicMock()
repository.daily_rows = AsyncMock(return_value=DailyRowsPage(total_count=0, rows=()))
with (
patch.object(team_endpoints, "prisma_client", prisma, create=True),
patch(
"litellm.proxy.management_endpoints.team_endpoints.get_user_object",
new=AsyncMock(return_value=user_info),
),
patch.object(common_daily_activity, "daily_activity_repository", return_value=repository),
patch("litellm.proxy.proxy_server.prisma_client", prisma),
patch("litellm.proxy.proxy_server.user_api_key_cache", MagicMock()),
patch("litellm.proxy.proxy_server.proxy_logging_obj", MagicMock()),
):
response = await team_endpoints.get_team_daily_activity(
team_ids="team-B",
start_date="2026-01-01",
end_date="2026-01-02",
api_key=requested_api_key,
user_api_key_dict=user,
)
scope = repository.daily_rows.await_args.args[0]
assert scope.api_keys == ()
sql, _params = build_where_clause(scope)
assert sql.endswith(" AND FALSE")
assert response.results == []
assert response.metadata.total_spend == 0

17
tests/unit/proxy/management_endpoints/test_common_daily_activity.py
View file

@ -10,6 +10,7 @@ from fastapi import HTTPException
import litellm.proxy.management_endpoints.common_daily_activity as common_daily_activity_module
from litellm.constants import USAGE_TOP_API_KEYS_DEFAULT
from litellm.proxy.management_endpoints.common_daily_activity import (
InvalidDateRange,
_is_user_agent_tag,
_ProxyDailyActivityReads,
_record_to_spend_metrics,
@ -18,6 +19,7 @@ from litellm.proxy.management_endpoints.common_daily_activity import (
daily_activity_scope,
get_api_key_metadata,
get_daily_activity,
raise_public,
update_metrics,
)
from litellm.proxy.management_endpoints.common_daily_activity import (
@ -2332,17 +2334,17 @@ async def test_get_api_key_metadata_resolves_session_key_via_spend_log_window():
def test_spend_logs_window_pads_min_minus_one_day_and_max_plus_two_days():
from litellm.proxy.management_endpoints.common_daily_activity import _spend_logs_window
from litellm.proxy.management_endpoints.common_daily_activity import spend_logs_window
window = _spend_logs_window({"2026-09-08", "2026-09-05", "not-a-date"})
window = spend_logs_window({"2026-09-08", "2026-09-05", "not-a-date"})
assert window == (datetime(2026, 9, 4), datetime(2026, 9, 10))
def test_spend_logs_window_is_none_when_no_date_parses():
from litellm.proxy.management_endpoints.common_daily_activity import _spend_logs_window
from litellm.proxy.management_endpoints.common_daily_activity import spend_logs_window
assert _spend_logs_window({"garbage", ""}) is None
assert spend_logs_window({"garbage", ""}) is None
@pytest.mark.asyncio
@ -2447,3 +2449,10 @@ async def test_get_api_key_metadata_does_not_recover_daily_spend_owner_for_activ
assert active_metadata.get("user_email") == "active-owner@example.com"
assert active_metadata.get("key_exists") is True
recovery_query_raw.assert_not_awaited()
def test_raise_public_maps_invalid_date_range_to_400() -> None:
with pytest.raises(HTTPException) as excinfo:
raise_public(InvalidDateRange(reason="Date range must be at most 400 days"))
assert excinfo.value.status_code == 400
assert excinfo.value.detail == {"error": "Date range must be at most 400 days"}

1261
tests/unit/proxy/management_endpoints/test_daily_activity_routes.py Normal file
View file

File diff suppressed because it is too large Load diff

182
tests/unit/proxy/management_endpoints/test_internal_user_endpoints.py
View file

@ -15,7 +15,6 @@ from fastapi.testclient import TestClient
from pytest_mock import MockerFixture
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
from litellm.proxy._types import (
LiteLLM_UserTableFiltered,
LitellmUserRoles,
@ -26,12 +25,10 @@ from litellm.proxy._types import (
UserAPIKeyAuth,
)
from litellm.proxy.management_endpoints.internal_user_endpoints import (
LiteLLM_UserTableWithKeyCount,
_authorize_user_list_request,
_resolve_org_filter_for_user_search,
_resolve_user_email_metadata,
_update_internal_user_params,
get_user_key_counts,
get_users,
new_user,
ui_view_users,
@ -2480,185 +2477,6 @@ async def test_get_user_daily_activity_rejects_service_account_caller(monkeypatc
mock_get_daily.assert_not_called()
@pytest.mark.asyncio
async def test_get_user_daily_activity_aggregated_rejects_service_account_caller(
monkeypatch,
):
"""
Same security regression as
test_get_user_daily_activity_rejects_service_account_caller, on the
aggregated route. Same shape, raw-SQL builder, same fix.
"""
from unittest.mock import AsyncMock, MagicMock
from fastapi import HTTPException
from litellm.proxy.management_endpoints.internal_user_endpoints import (
get_user_daily_activity_aggregated,
)
mock_prisma_client = MagicMock()
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client)
mock_get_daily_agg = AsyncMock()
monkeypatch.setattr(
"litellm.proxy.management_endpoints.internal_user_endpoints.get_daily_activity_aggregated",
mock_get_daily_agg,
)
service_account_key = UserAPIKeyAuth(
user_id=None,
user_role=LitellmUserRoles.INTERNAL_USER,
)
with pytest.raises(HTTPException) as exc_info:
await get_user_daily_activity_aggregated(
start_date="2025-01-01",
end_date="2025-01-31",
model=None,
api_key=None,
user_id=None,
timezone=None,
user_api_key_dict=service_account_key,
)
assert exc_info.value.status_code == 403
assert "Service-account keys" in str(exc_info.value.detail)
mock_get_daily_agg.assert_not_called()
@pytest.mark.asyncio
@pytest.mark.parametrize("include_current_utc_day", [False, True])
async def test_get_user_daily_activity_aggregated_admin_global_view(monkeypatch, include_current_utc_day):
"""
Test that admin users can call the aggregated endpoint without a user_id
to get a global view. Also verifies that the correct arguments are forwarded
to the underlying get_daily_activity_aggregated helper.
"""
from unittest.mock import AsyncMock, MagicMock
from litellm.proxy.management_endpoints.internal_user_endpoints import (
get_user_daily_activity_aggregated,
)
# Mock the prisma client
mock_prisma_client = MagicMock()
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client)
# Mock the downstream helper so we don't need a real DB
mock_response = MagicMock()
mock_get_daily_agg = AsyncMock(return_value=mock_response)
monkeypatch.setattr(
"litellm.proxy.management_endpoints.internal_user_endpoints.get_daily_activity_aggregated",
mock_get_daily_agg,
)
# Admin caller
admin_key_dict = UserAPIKeyAuth(
user_id="admin-user-001",
user_role=LitellmUserRoles.PROXY_ADMIN,
)
# Admin calls without user_id → global view (entity_id=None)
result = await get_user_daily_activity_aggregated(
start_date="2025-02-01",
end_date="2025-02-28",
model="gpt-4",
api_key=None,
user_id=None,
timezone=480,
include_current_utc_day=include_current_utc_day,
user_api_key_dict=admin_key_dict,
)
assert result is mock_response
# Verify the helper was called with the right parameters
mock_get_daily_agg.assert_called_once()
repository, scope = mock_get_daily_agg.call_args.args
assert repository is not None
assert scope.table.value == "litellm_dailyuserspend"
assert scope.entity_id_field == "user_id"
assert scope.entity_ids is None
assert scope.start_date == "2025-02-01"
assert scope.end_date == "2025-02-28"
assert scope.model == "gpt-4"
assert scope.api_keys is None
assert scope.timezone_offset_minutes == 480
assert scope.include_current_utc_day is include_current_utc_day
@pytest.mark.asyncio
async def test_get_user_daily_activity_aggregated_non_admin_cannot_view_other_users(
monkeypatch,
):
"""
Same scoping contract as
test_get_user_daily_activity_non_admin_cannot_view_other_users, on the
aggregated route. Non-admins reach this handler now that the route is in
self_managed_routes, so the 403-on-mismatch and default-to-self behaviour
has to hold here too: opening the route must not widen access.
"""
from unittest.mock import AsyncMock, MagicMock, patch
from fastapi import HTTPException
from litellm.proxy.management_endpoints.internal_user_endpoints import (
get_user_daily_activity_aggregated,
)
mock_prisma_client = MagicMock()
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client)
non_admin_key_dict = UserAPIKeyAuth(
user_id="regular-user-123",
user_role=LitellmUserRoles.INTERNAL_USER,
)
# Case 1: Non-admin targets another user's data — 403, helper never reached
with patch(
"litellm.proxy.management_endpoints.internal_user_endpoints.get_daily_activity_aggregated",
new_callable=AsyncMock,
) as mock_get_daily_agg:
with pytest.raises(HTTPException) as exc_info:
await get_user_daily_activity_aggregated(
start_date="2025-01-01",
end_date="2025-01-31",
model=None,
api_key=None,
user_id="other-user-456",
timezone=None,
user_api_key_dict=non_admin_key_dict,
)
assert exc_info.value.status_code == 403
assert "Non-admin users can only view their own spend data" in str(exc_info.value.detail)
mock_get_daily_agg.assert_not_called()
# Case 2: Non-admin omits user_id — scoped to their own user_id, not global
mock_response = MagicMock()
with patch(
"litellm.proxy.management_endpoints.internal_user_endpoints.get_daily_activity_aggregated",
new_callable=AsyncMock,
return_value=mock_response,
) as mock_get_daily_agg:
result = await get_user_daily_activity_aggregated(
start_date="2025-01-01",
end_date="2025-01-31",
model=None,
api_key=None,
user_id=None,
timezone=None,
user_api_key_dict=non_admin_key_dict,
)
assert result is mock_response
mock_get_daily_agg.assert_called_once()
repository, scope = mock_get_daily_agg.call_args.args
assert repository is not None
assert scope.entity_ids == ("regular-user-123",)
@pytest.mark.asyncio
async def test_delete_user_cleans_up_created_by_invitation_links(mocker):
"""

2
tests/unit/proxy/management_endpoints/test_organization_endpoints.py
View file

@ -1292,7 +1292,7 @@ async def test_get_organization_daily_activity_non_admin_without_org_admin_role_
)
assert get_daily_activity_mock.call_args.kwargs["entity_id"] == []
assert org_table_find_many.call_args.kwargs["where"] == {"organization_id": {"in": []}}
org_table_find_many.assert_not_awaited()
@pytest.mark.asyncio

125
tests/unit/proxy/management_endpoints/test_team_endpoints.py
View file

@ -1,10 +1,10 @@
import asyncio
import json
from collections.abc import Sequence
from contextlib import AbstractContextManager, asynccontextmanager, contextmanager
from dataclasses import dataclass
from datetime import datetime, timezone
from types import SimpleNamespace
from collections.abc import Sequence
from typing import Final, Optional, cast
from unittest.mock import AsyncMock, MagicMock, PropertyMock, call, patch
@ -14602,129 +14602,6 @@ async def test_new_team_batch_enqueued_token_limit_rejected_for_non_admin():
assert "on a team" in str(exc.value.message)
@pytest.mark.asyncio
async def test_get_team_daily_activity_aggregated_scopes_and_flags(mock_db_client):
"""The aggregated endpoint must apply the same non-admin key scoping as the
paginated one and request the per-team entity breakdown with the caller's
timezone, so the Team Usage UI gets every day in one response."""
from litellm.proxy.management_endpoints.team_endpoints import (
get_team_daily_activity_aggregated,
)
user_id = "test_user_123"
team_id = "test_team_456"
user_api_key_dict = UserAPIKeyAuth(
user_id=user_id, user_role=LitellmUserRoles.INTERNAL_USER
)
mock_user_info = LiteLLM_UserTable(
user_id=user_id,
teams=[team_id],
max_budget=1000.0,
spend=0.0,
user_email="test@example.com",
user_role="internal_user",
)
mock_team_member = Member(user_id=user_id, role="user")
mock_team = MagicMock(spec=LiteLLM_TeamTable)
mock_team.team_id = team_id
mock_team.team_alias = "Test Team"
mock_team.members_with_roles = [mock_team_member]
mock_team.model_dump.return_value = {
"team_id": team_id,
"team_alias": "Test Team",
"members_with_roles": [{"user_id": user_id, "role": "user"}],
}
user_api_key_1 = MagicMock()
user_api_key_1.token = "user_key_1"
mock_db_client.db.litellm_teamtable.find_many = AsyncMock(return_value=[mock_team])
mock_db_client.db.litellm_verificationtoken.find_many = AsyncMock(
return_value=[user_api_key_1]
)
with patch(
"litellm.proxy.management_endpoints.team_endpoints.get_user_object",
new_callable=AsyncMock,
) as mock_get_user_object:
mock_get_user_object.return_value = mock_user_info
with patch(
"litellm.proxy.management_endpoints.team_endpoints.get_daily_activity_aggregated",
new_callable=AsyncMock,
) as mock_aggregated:
mock_aggregated.return_value = MagicMock()
await get_team_daily_activity_aggregated(
team_ids=team_id,
start_date="2024-01-01",
end_date="2024-01-31",
model=None,
api_key=None,
exclude_team_ids=None,
timezone=480,
user_api_key_dict=user_api_key_dict,
)
mock_aggregated.assert_called_once()
repository, scope = mock_aggregated.call_args.args
call_kwargs = mock_aggregated.call_args.kwargs
assert repository is not None
assert scope.api_keys == ("user_key_1",)
assert scope.entity_ids == (team_id,)
assert call_kwargs["entity_metadata_field"] == {
team_id: {"team_alias": "Test Team"}
}
assert call_kwargs["include_entity_breakdown"] is True
assert scope.timezone_offset_minutes == 480
assert scope.table.value == "litellm_dailyteamspend"
@pytest.mark.asyncio
@pytest.mark.parametrize(
"start_date,end_date,expected_error",
[
("2020-01-01", "2026-12-31", "at most 400 days"),
("0000-01-01", "9999-12-31", "valid YYYY-MM-DD"),
("2024-06-01", "2024-01-01", "on or after"),
("not-a-date", "2024-01-31", "valid YYYY-MM-DD"),
(None, "2024-01-31", "start_date and end_date"),
],
)
async def test_get_team_daily_activity_aggregated_rejects_bad_ranges(
mock_db_client, start_date, end_date, expected_error
):
"""The aggregated endpoint has no pagination bounding its work, so an
unbounded or malformed range must 400 before any query runs."""
from litellm.proxy.management_endpoints.team_endpoints import (
get_team_daily_activity_aggregated,
)
with patch(
"litellm.proxy.management_endpoints.team_endpoints.get_daily_activity_aggregated",
new_callable=AsyncMock,
) as mock_aggregated:
with pytest.raises(HTTPException) as exc_info:
await get_team_daily_activity_aggregated(
team_ids=None,
start_date=start_date,
end_date=end_date,
model=None,
api_key=None,
exclude_team_ids=None,
timezone=None,
user_api_key_dict=UserAPIKeyAuth(
user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN
),
)
assert exc_info.value.status_code == 400
assert expected_error in str(exc_info.value.detail)
mock_aggregated.assert_not_called()
def _wire_new_team_prisma(mock_db_client):
mock_db_client.jsonify_team_object = lambda db_data: db_data
mock_db_client.get_data = AsyncMock(return_value=None)

2041
ui/litellm-dashboard/src/lib/http/schema.d.ts generated vendored
View file

File diff suppressed because it is too large Load diff