Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/management_endpoints/internal_user_endpoints.py: 48%
949 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""
2Internal User Management Endpoints
5These are members of a Team on LiteLLM
7/user/new
8/user/update
9/user/bulk_update
10/user/delete
11/user/info
12/user/list
13"""
15import asyncio
16import json
17import traceback
18from collections.abc import Awaitable, Mapping, Sequence
19from datetime import datetime, timezone
20from types import MappingProxyType
21from typing import Any, Final, Literal, Protocol, cast, overload
23import fastapi
24from fastapi import APIRouter, Depends, Header, HTTPException, Request, status
25from pydantic import TypeAdapter, ValidationError
26from typing_extensions import ReadOnly, TypedDict
28import litellm
29from litellm._logging import verbose_proxy_logger
30from litellm._uuid import uuid
31from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
32from litellm.proxy._types import *
33from litellm.proxy.auth.auth_checks import (
34 delete_cache_key_objects,
35 get_jwt_key_mapping_cache_keys_for_tokens,
36 get_team_object,
37 get_user_object,
38)
39from litellm.proxy.auth.password_policy import (
40 validate_password_not_breached,
41 validate_password_policy,
42 validate_passwords_bulk,
43)
44from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
45from litellm.proxy.common_utils.auth_cache_invalidation_pubsub import evict_and_broadcast
46from litellm.proxy.common_utils.user_api_key_cache import (
47 object_permission_cache_key,
48 user_object_permission_id_cache_key,
49)
50from litellm.proxy.db.exception_handler import PrismaDBExceptionHandler
51from litellm.proxy.hooks.key_management_event_hooks import KeyManagementEventHooks
52from litellm.proxy.hooks.model_max_budget_limiter import build_model_max_budget_usage
53from litellm.proxy.hooks.user_management_event_hooks import UserManagementEventHooks
54from litellm.proxy.management_endpoints.common_daily_activity import (
55 DailySpendRecord,
56 get_daily_activity,
57 get_daily_activity_aggregated,
58)
59from litellm.proxy.management_endpoints.common_utils import (
60 _is_user_team_admin,
61 _user_has_admin_view,
62 require_caller_user_id_for_non_admin,
63 validate_budget_duration,
64 validate_finite_spend,
65)
66from litellm.proxy.management_endpoints.key_management_endpoints import (
67 _check_permissions_caller_permission,
68 generate_key_helper_fn,
69 prepare_metadata_fields,
70)
71from litellm.proxy.management_helpers.object_permission_utils import (
72 _set_object_permission,
73 handle_update_object_permission_common,
74)
75from litellm.proxy.management_helpers.utils import management_endpoint_wrapper
76from litellm.proxy.utils import handle_exception_on_proxy, hash_password
77from litellm.repositories.organization_repository import OrganizationRepository
78from litellm.repositories.prisma_protocols import TableActions
79from litellm.repositories.table_repositories import (
80 InvitationLinkRepository,
81 OrganizationMembershipRepository,
82 TeamMembershipRepository,
83)
84from litellm.repositories.team_repository import TeamRepository
85from litellm.repositories.user_repository import UserRepository
86from litellm.repositories.verification_token_repository import (
87 VerificationTokenRepository,
88)
89from litellm.types.proxy.management_endpoints.common_daily_activity import (
90 SpendAnalyticsPaginatedResponse,
91)
92from litellm.types.proxy.management_endpoints.internal_user_endpoints import (
93 BulkUpdateUserRequest,
94 BulkUpdateUserResponse,
95 UserListResponse,
96 UserSearchWhere,
97 UserUpdateResult,
98)
99from litellm.types.proxy.management_endpoints.scim_v2 import (
100 SCIM_ENTERPRISE_METADATA_KEY,
101 SCIM_ENTITLEMENTS_METADATA_KEY,
102 SCIM_ROLES_METADATA_KEY,
103)
104from litellm.types.utils import BudgetConfig
106if TYPE_CHECKING: 106 ↛ 107line 106 didn't jump to line 107 because the condition on line 106 was never true
107 from prisma import models as prisma_models
108 from prisma import types as prisma_types
110 from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
111 from litellm.proxy.proxy_server import PrismaClient
112 from litellm.proxy.utils import ProxyLogging
114router: Final = APIRouter()
115_USER_MODEL_BUDGET_ADAPTER: Final = TypeAdapter(dict[str, float | BudgetConfig])
116_USER_BUDGET_CACHE_INVALIDATION_BATCH_SIZE: Final = 50
117_USER_BUDGET_CACHE_FIELDS: Final = frozenset({"max_budget", "model_max_budget"})
120def _user_table(
121 prisma_client: "PrismaClient | None",
122) -> "TableActions[prisma_models.LiteLLM_UserTable]":
123 user_table: Final[TableActions[prisma_models.LiteLLM_UserTable]] = UserRepository(prisma_client).table
124 return user_table
127def _team_table(
128 prisma_client: "PrismaClient | None",
129) -> "TableActions[prisma_models.LiteLLM_TeamTable]":
130 team_table: Final[TableActions[prisma_models.LiteLLM_TeamTable]] = TeamRepository(prisma_client).table
131 return team_table
134def _verification_token_table(
135 prisma_client: "PrismaClient | None",
136) -> "TableActions[prisma_models.LiteLLM_VerificationToken]":
137 token_table: Final[TableActions[prisma_models.LiteLLM_VerificationToken]] = VerificationTokenRepository(
138 prisma_client
139 ).table
140 return token_table
143class _UserIdInFilter(TypedDict):
144 user_id: ReadOnly[Mapping[str, Sequence[str]]]
147def _organization_membership_table(
148 prisma_client: "PrismaClient | None",
149) -> "TableActions[prisma_models.LiteLLM_OrganizationMembership]":
150 membership_table: Final[TableActions[prisma_models.LiteLLM_OrganizationMembership]] = (
151 OrganizationMembershipRepository(prisma_client).table
152 )
153 return membership_table
156def _invitation_link_table(
157 prisma_client: "PrismaClient | None",
158) -> "TableActions[prisma_models.LiteLLM_InvitationLink]":
159 invitation_table: Final[TableActions[prisma_models.LiteLLM_InvitationLink]] = InvitationLinkRepository(
160 prisma_client
161 ).table
162 return invitation_table
165def _organization_table(
166 prisma_client: "PrismaClient | None",
167) -> "TableActions[prisma_models.LiteLLM_OrganizationTable]":
168 organization_table: Final[TableActions[prisma_models.LiteLLM_OrganizationTable]] = OrganizationRepository(
169 prisma_client
170 ).table
171 return organization_table
174def _team_membership_table(
175 prisma_client: "PrismaClient | None",
176) -> "TableActions[prisma_models.LiteLLM_TeamMembership]":
177 team_membership_table: Final[TableActions[prisma_models.LiteLLM_TeamMembership]] = TeamMembershipRepository(
178 prisma_client
179 ).table
180 return team_membership_table
183async def _hash_password_in_dict(
184 data: dict, general_settings: Mapping[str, object], password_prevalidated: bool = False
185) -> None:
186 """Validate and hash password field in-place if present.
188 ``password_prevalidated`` skips the policy checks for callers that already
189 validated the password (the bulk path screens its whole batch upfront).
191 An admin-set password is known to whoever set it, so the user is also
192 flagged for a forced password change at next login."""
193 if "password" in data and data["password"] is not None:
194 if not password_prevalidated:
195 validate_password_policy(data["password"], general_settings)
196 await validate_password_not_breached(data["password"], general_settings)
197 data["password"] = hash_password(data["password"])
198 data["password_reset_required"] = True
199 data["last_breach_check_at"] = None
202def _strip_password_from_response(response) -> None:
203 """Strip password from API response (handles dicts, nested dicts, and Prisma models)."""
204 if isinstance(response, dict):
205 response.pop("password", None)
206 if isinstance(response.get("data"), dict):
207 response["data"].pop("password", None)
208 elif hasattr(response.get("data"), "__dict__"):
209 response["data"].__dict__.pop("password", None)
212def _update_internal_new_user_params(data_json: dict, data: NewUserRequest) -> dict:
213 if "user_id" in data_json and data_json["user_id"] is None:
214 data_json["user_id"] = str(uuid.uuid4())
216 auto_create_key: Final = data_json.pop("auto_create_key", True)
218 if auto_create_key is False:
219 data_json["table_name"] = "user" # only create a user, don't create key if 'auto_create_key' set to False
221 if litellm.default_internal_user_params and (
222 data.user_role != LitellmUserRoles.PROXY_ADMIN.value and data.user_role != LitellmUserRoles.PROXY_ADMIN
223 ):
224 for key, value in litellm.default_internal_user_params.items():
225 if key == "available_teams": 225 ↛ 226line 225 didn't jump to line 226 because the condition on line 225 was never true
226 continue
227 elif (
228 key not in data_json
229 or data_json[key] is None
230 or key == "models"
231 and isinstance(data_json[key], list)
232 and len(data_json[key]) == 0
233 ):
234 data_json[key] = value
236 ## INTERNAL USER ROLE ONLY DEFAULT PARAMS ##
237 if data.user_role is not None and data.user_role == LitellmUserRoles.INTERNAL_USER.value:
238 if litellm.max_internal_user_budget is not None and data_json.get("max_budget") is None: 238 ↛ 239line 238 didn't jump to line 239 because the condition on line 238 was never true
239 data_json["max_budget"] = litellm.max_internal_user_budget
241 if litellm.internal_user_budget_duration is not None and data_json.get("budget_duration") is None: 241 ↛ 242line 241 didn't jump to line 242 because the condition on line 241 was never true
242 data_json["budget_duration"] = litellm.internal_user_budget_duration
244 data_json.pop("teams", None) # handled separately
245 return data_json
248async def _check_duplicate_user_field(
249 field_name: str,
250 field_value: str | None,
251 prisma_client: "PrismaClient | None",
252 *,
253 case_insensitive: bool = False,
254 label: str | None = None,
255) -> None:
256 """
257 Helper function to check if a field already exists in the user table.
259 Args:
260 field_name (str): Database field name to check.
261 field_value (Optional[str]): Value to check for duplicates.
262 prisma_client (Any): Database client instance.
263 case_insensitive (bool): Whether to use case-insensitive comparison.
264 label (Optional[str]): Human readable label for error messages.
266 Raises:
267 Exception: If database is not connected.
268 HTTPException: If a user with the given field value already exists.
269 """
270 if field_value:
271 if prisma_client is None: 271 ↛ 272line 271 didn't jump to line 272 because the condition on line 271 was never true
272 raise Exception("Database not connected")
274 value: Final = field_value.strip()
275 where_clause: Final = {field_name: {"equals": value}}
276 if case_insensitive: 276 ↛ 277line 276 didn't jump to line 277 because the condition on line 276 was never true
277 where_clause[field_name]["mode"] = "insensitive"
279 existing_user: Final[object] = await UserRepository(prisma_client).table.find_first(where=where_clause)
281 if existing_user is not None:
282 existing_value: Final = getattr(existing_user, field_name, value)
283 error_label: Final = label or field_name
284 raise HTTPException(
285 status_code=409,
286 detail={"error": f"User with {error_label} {existing_value} already exists"},
287 )
290async def _check_duplicate_user_email(user_email: str | None, prisma_client: "PrismaClient | None") -> None:
291 """
292 Helper function to check if a user email already exists in the database.
293 """
294 await _check_duplicate_user_field(
295 field_name="user_email",
296 field_value=user_email,
297 prisma_client=prisma_client,
298 case_insensitive=True,
299 label="email",
300 )
303async def _check_duplicate_user_id(user_id: str | None, prisma_client: "PrismaClient | None") -> None:
304 """
305 Helper function to check if a user id already exists in the database.
306 """
307 await _check_duplicate_user_field(
308 field_name="user_id",
309 field_value=user_id,
310 prisma_client=prisma_client,
311 label="id",
312 )
315async def _add_user_to_organizations(
316 user_id: str,
317 organizations: list[str],
318 prisma_client: "PrismaClient",
319 user_api_key_dict: UserAPIKeyAuth,
320):
321 """
322 Add a user to organizations
323 """
324 from litellm.proxy.management_endpoints.organization_endpoints import (
325 organization_member_add,
326 )
328 tasks: Final[list[Awaitable[object]]] = []
329 for organization_id in organizations:
330 tasks.append(
331 organization_member_add(
332 data=OrganizationMemberAddRequest(
333 organization_id=organization_id,
334 member=[
335 OrgMember(
336 user_id=user_id,
337 role=LitellmUserRoles.INTERNAL_USER,
338 )
339 ],
340 ),
341 http_request=Request(
342 scope={"type": "http", "path": "/user/new"},
343 ),
344 user_api_key_dict=user_api_key_dict,
345 )
346 )
347 await asyncio.gather(*tasks, return_exceptions=True)
350async def _add_user_to_team(
351 user_id: str,
352 team_id: str,
353 user_api_key_dict: UserAPIKeyAuth,
354 user_email: str | None = None,
355 max_budget_in_team: float | None = None,
356 user_role: Literal["user", "admin"] = "user",
357):
358 from litellm.proxy.management_endpoints.team_endpoints import team_member_add
360 try:
361 await team_member_add(
362 data=TeamMemberAddRequest(
363 team_id=team_id,
364 member=Member(
365 user_id=user_id,
366 role=user_role,
367 user_email=user_email,
368 ),
369 max_budget_in_team=max_budget_in_team,
370 ),
371 user_api_key_dict=user_api_key_dict,
372 )
373 except HTTPException as e:
374 if e.status_code == 400 and ("already exists" in str(e) or "doesn't exist" in str(e)): 374 ↛ 375line 374 didn't jump to line 375 because the condition on line 374 was never true
375 verbose_proxy_logger.debug(
376 "litellm.proxy.management_endpoints.internal_user_endpoints.new_user(): User already exists in team - %s",
377 e,
378 )
379 else:
380 verbose_proxy_logger.error(
381 "litellm.proxy.management_endpoints.internal_user_endpoints._add_user_to_team(): "
382 "failed to add user %s to team %s - %s",
383 user_id,
384 team_id,
385 str(e),
386 )
387 except Exception as e:
388 if (
389 "already exists" in str(e)
390 or "doesn't exist" in str(e)
391 or isinstance(e, ProxyException)
392 and ProxyErrorTypes.team_member_already_in_team in e.type
393 ):
394 verbose_proxy_logger.debug(
395 "litellm.proxy.management_endpoints.internal_user_endpoints.new_user(): User already exists in team - %s",
396 e,
397 )
398 else:
399 verbose_proxy_logger.error(
400 "litellm.proxy.management_endpoints.internal_user_endpoints._add_user_to_team(): "
401 "failed to add user %s to team %s - %s",
402 user_id,
403 team_id,
404 str(e),
405 )
406 raise e
409def check_if_default_team_set() -> list[str] | list[NewUserRequestTeam] | None:
410 if litellm.default_internal_user_params is None: 410 ↛ 411line 410 didn't jump to line 411 because the condition on line 410 was never true
411 return None
412 teams: Final = litellm.default_internal_user_params.get("teams")
413 if teams is not None:
414 if all(isinstance(team, str) for team in teams): 414 ↛ 416line 414 didn't jump to line 416 because the condition on line 414 was always true
415 return teams
416 elif all(isinstance(team, dict) for team in teams):
417 return [
418 NewUserRequestTeam(
419 team_id=team.get("team_id"),
420 max_budget_in_team=team.get("max_budget_in_team"),
421 user_role=team.get("user_role", "user"),
422 )
423 for team in teams
424 ]
425 else:
426 verbose_proxy_logger.error(
427 "Invalid team type in default internal user params: %s",
428 teams,
429 )
430 return None
433async def add_new_user_to_default_team(
434 user_id: str,
435 user_email: str | None,
436 user_api_key_dict: UserAPIKeyAuth,
437 teams: list[str] | list[NewUserRequestTeam],
438 prisma_client: "PrismaClient",
439):
440 tasks: Final[list[Awaitable[object]]] = []
441 for team in teams:
442 user_role: Literal["user", "admin"] = "user"
443 max_budget_in_team: float | None = None
444 if isinstance(team, str):
445 team_id = team
446 elif isinstance(team, NewUserRequestTeam): 446 ↛ 451line 446 didn't jump to line 451 because the condition on line 446 was always true
447 team_id = team.team_id
448 user_role = team.user_role
449 max_budget_in_team = team.max_budget_in_team
450 else:
451 raise ValueError(f"Invalid team type: {type(team)}")
453 tasks.append(
454 _add_user_to_team(
455 user_id=user_id,
456 team_id=team_id,
457 user_email=user_email,
458 user_api_key_dict=user_api_key_dict,
459 max_budget_in_team=max_budget_in_team,
460 user_role=user_role,
461 )
462 )
463 await asyncio.gather(*tasks, return_exceptions=True)
466async def _fetch_user_team_ids(user_id: str, prisma_client: "PrismaClient") -> tuple[str, ...]:
467 user_row: Final = await _user_table(prisma_client).find_unique(where={"user_id": user_id})
468 return tuple(user_row.teams) if user_row is not None else ()
471@router.post(
472 "/user/new",
473 tags=["Internal User management"],
474 dependencies=[Depends(user_api_key_auth)],
475 response_model=NewUserResponse,
476)
477@management_endpoint_wrapper
478async def new_user(
479 data: NewUserRequest,
480 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
481):
482 """
483 Use this to create a new INTERNAL user with a budget.
484 Internal Users can access LiteLLM Admin UI to make keys, request access to models.
485 This creates a new user and generates a new api key for the new user. The new api key is returned.
487 Returns user id, budget + new key.
489 Parameters:
490 - user_id: Optional[str] - Specify a user id. If not set, a unique id will be generated.
491 - user_alias: Optional[str] - A descriptive name for you to know who this user id refers to.
492 - teams: Optional[list] - specify a list of team id's a user belongs to.
493 - user_email: Optional[str] - Specify a user email.
494 - send_invite_email: Optional[bool] - Specify if an invite email should be sent.
495 - user_role: Optional[str] - Specify a user role - "proxy_admin", "proxy_admin_viewer", "internal_user", "internal_user_viewer", "team", "customer". Info about each role here: `https://github.com/BerriAI/litellm/litellm/proxy/_types.py#L20`
496 - max_budget: Optional[float] - Specify max budget for a given user.
497 - budget_duration: Optional[str] - Budget is reset at the end of specified duration. If not set, budget is never reset. You can set duration as seconds ("30s"), minutes ("30m"), hours ("30h"), days ("30d"), months ("1mo").
498 - models: Optional[list] - Model_name's a user is allowed to call. (if empty, key is allowed to call all models). Set to ['no-default-models'] to block all model access. Restricting user to only team-based model access.
499 - tpm_limit: Optional[int] - Specify tpm limit for a given user (Tokens per minute)
500 - rpm_limit: Optional[int] - Specify rpm limit for a given user (Requests per minute)
501 - auto_create_key: bool - Default=True. Flag used for returning a key as part of the /user/new response
502 - aliases: Optional[dict] - Model aliases for the user - [Docs](https://litellm.vercel.app/docs/proxy/virtual_keys#model-aliases)
503 - config: Optional[dict] - [DEPRECATED PARAM] User-specific config.
504 - allowed_cache_controls: Optional[list] - List of allowed cache control values. Example - ["no-cache", "no-store"]. See all values - https://docs.litellm.ai/docs/proxy/caching#turn-on--off-caching-per-request-
505 - blocked: Optional[bool] - [Not Implemented Yet] Whether the user is blocked.
506 - guardrails: Optional[List[str]] - [Not Implemented Yet] List of active guardrails for the user
507 - policies: Optional[List[str]] - List of policy names to apply to the user. Policies define guardrails, conditions, and inheritance rules.
508 - permissions: Optional[dict] - [Not Implemented Yet] User-specific permissions, eg. turning off pii masking.
509 - metadata: Optional[dict] - Metadata for user, store information for user. Example metadata = {"team": "core-infra", "app": "app2", "email": "ishaan@berri.ai" }
510 - max_parallel_requests: Optional[int] - Rate limit a user based on the number of parallel requests. Raises 429 error, if user's parallel requests > x.
511 - model_max_budget: Optional[dict] - Model-specific max budget for user. [Docs](https://docs.litellm.ai/docs/proxy/users#add-model-specific-budgets-to-keys)
512 - budget_fallbacks: Optional[Dict[str, List[str]]] - Per-model fallback chain tried in order when that model's own `model_max_budget` is exceeded, e.g. {"gpt-4o": ["gpt-4o-mini"]}.
513 - model_rpm_limit: Optional[float] - Model-specific rpm limit for user. [Docs](https://docs.litellm.ai/docs/proxy/users#add-model-specific-limits-to-keys)
514 - mcp_rpm_limit: Optional[dict] - Per-MCP-server rpm limit, keyed by MCP server name {"github": 100, "slack": 200}. Enforced for keys and teams only; values set on a user are stored but not enforced per user.
515 - tag_rpm_limit: Optional[dict] - Per-request-tag rpm limit, keyed by request tag {"cell-1": 1000, "cell-2": 500}. Enforced for keys only; values set on a user are stored but not enforced per user.
516 - model_tpm_limit: Optional[float] - Model-specific tpm limit for user. [Docs](https://docs.litellm.ai/docs/proxy/users#add-model-specific-limits-to-keys)
517 - spend: Optional[float] - Amount spent by user. Default is 0. Will be updated by proxy whenever user is used. You can set duration as seconds ("30s"), minutes ("30m"), hours ("30h"), days ("30d"), months ("1mo").
518 - agent_id: Optional[str] - The agent id associated with the user.
519 - team_id: Optional[str] - [DEPRECATED PARAM] The team id of the user. Default is None.
520 - duration: Optional[str] - Duration for the key auto-created on `/user/new`. Default is None.
521 - key_alias: Optional[str] - Alias for the key auto-created on `/user/new`. Default is None.
522 - sso_user_id: Optional[str] - The id of the user in the SSO provider.
523 - object_permission: Optional[LiteLLM_ObjectPermissionBase] - internal user-specific object permission. Example - {"vector_stores": ["vector_store_1"], "mcp_servers": ["github"], "mcp_tool_permissions": {"github": ["list_issues"]}}. The MCP grants act as a ceiling on every key this user holds. IF null or {} then no object permission.
524 - prompts: Optional[List[str]] - List of allowed prompts for the user. If specified, the user will only be able to use these specific prompts.
525 - organizations: List[str] - List of organization id's the user is a member of
526 - budget_limits: Optional[list] - List of concurrent budget windows for the user. Each window specifies a budget_limit, time_period, and optional budget_duration. Example - [{"budget_limit": 10.0, "time_period": "1d"}, {"budget_limit": 50.0, "time_period": "7d"}].
527 - password: Optional[str] - Not supported; any value is rejected with a 422. Users set their own password through an invitation link (POST /invitation/new).
528 Returns:
529 - key: (str) The generated api key for the user
530 - expires: (datetime) Datetime object for when key expires.
531 - user_id: (str) Unique user id - used for tracking spend across multiple keys for same user id.
532 - max_budget: (float|None) Max budget for given user.
534 Usage Example
536 ```shell
537 curl -X POST "http://localhost:4000/user/new" \
538 -H "Content-Type: application/json" \
539 -H "Authorization: Bearer sk-1234" \
540 -d '{
541 "username": "new_user",
542 "email": "new_user@example.com"
543 }'
544 ```
545 """
546 try:
547 from litellm.proxy.proxy_server import _license_check, prisma_client
549 if prisma_client is None: 549 ↛ 550line 549 didn't jump to line 550 because the condition on line 549 was never true
550 raise HTTPException(status_code=400, detail=CommonProxyErrors.db_not_connected_error.value)
552 if prisma_client is None: 552 ↛ 553line 552 didn't jump to line 553 because the condition on line 552 was never true
553 raise HTTPException(
554 status_code=500,
555 detail=CommonProxyErrors.db_not_connected_error.value,
556 )
557 validate_budget_duration(data.budget_duration)
559 # Check for duplicate user_id or email
560 await _check_duplicate_user_id(data.user_id, prisma_client)
561 await _check_duplicate_user_email(data.user_email, prisma_client)
563 # Check if license is over limit
564 billable_users: Final = await UserRepository(prisma_client).count_billable_users()
565 if billable_users and _license_check.is_over_limit(total_users=billable_users): 565 ↛ 566line 565 didn't jump to line 566 because the condition on line 565 was never true
566 raise HTTPException(
567 status_code=403,
568 detail="License is over limit. Please contact support@berri.ai to upgrade your license.",
569 )
571 # Only proxy admins can create administrative users
572 # Check if user_api_key_dict is actually a UserAPIKeyAuth instance (not a Depends object)
573 # This can happen when the function is called directly in tests
574 if ( 574 ↛ 579line 574 didn't jump to line 579 because the condition on line 574 was never true
575 data.user_role in [LitellmUserRoles.PROXY_ADMIN, LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY]
576 and isinstance(user_api_key_dict, UserAPIKeyAuth)
577 and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN
578 ):
579 raise HTTPException(
580 status_code=403,
581 detail=f"Only proxy admins can create administrative users (proxy_admin, proxy_admin_viewer). Attempted to create user with role: {data.user_role}. Your role: {user_api_key_dict.user_role}",
582 )
584 _check_permissions_caller_permission(
585 data=data,
586 user_api_key_dict=user_api_key_dict,
587 )
589 data_json = data.json()
590 data_json = _update_internal_new_user_params(data_json, data)
591 # Persist the requested grants as their own row and link it, mirroring key/team creation.
592 # generate_key_helper_fn only forwards object_permission_id, so without this the entitlement
593 # the caller sent would be dropped on the floor.
594 data_json = await _set_object_permission(data_json=data_json, prisma_client=prisma_client)
595 data_json.pop("password", None)
596 teams = data.teams
597 if teams is None:
598 teams = check_if_default_team_set()
599 organization_ids: Final = cast(list[str] | None, data_json.pop("organizations", None))
601 response: Final = await generate_key_helper_fn(request_type="user", **data_json, llm_router=None)
602 # Admin UI Logic
603 # Add User to Team and Organization
604 # if team_id passed add this user to the team
605 _team_id: Final = data_json.get("team_id", None)
606 if _team_id is not None: 606 ↛ 607line 606 didn't jump to line 607 because the condition on line 606 was never true
607 await _add_user_to_team(
608 user_id=cast(str, response.get("user_id")),
609 team_id=_team_id,
610 user_api_key_dict=user_api_key_dict,
611 user_email=data.user_email,
612 max_budget_in_team=None,
613 user_role="user",
614 )
615 elif teams is not None:
616 await add_new_user_to_default_team(
617 user_id=cast(str, response.get("user_id")),
618 user_email=data.user_email,
619 user_api_key_dict=user_api_key_dict,
620 teams=teams,
621 prisma_client=prisma_client,
622 )
624 user_id: Final = cast(str | None, response.get("user_id", None))
625 attached_team_ids: Final = (
626 await _fetch_user_team_ids(user_id=user_id, prisma_client=prisma_client)
627 if user_id is not None and (_team_id is not None or teams is not None)
628 else None
629 )
631 if organization_ids is not None and user_id is not None:
632 await _add_user_to_organizations(
633 user_id=user_id,
634 organizations=organization_ids,
635 prisma_client=prisma_client,
636 user_api_key_dict=user_api_key_dict,
637 )
639 special_keys: Final = ["token", "token_id"]
640 response_dict: Final = {}
641 for key, value in response.items():
642 if key in NewUserResponse.model_fields and key not in special_keys:
643 response_dict[key] = value
645 response_dict["key"] = response.get("token", "")
646 if attached_team_ids is not None:
647 response_dict["teams"] = list(attached_team_ids)
649 new_user_response: Final = NewUserResponse.model_validate(response_dict)
651 #########################################################
652 ########## USER CREATED HOOK ################
653 #########################################################
654 asyncio.create_task(
655 UserManagementEventHooks.async_user_created_hook(
656 data=data,
657 response=new_user_response,
658 user_api_key_dict=user_api_key_dict,
659 )
660 )
661 #########################################################
662 ########## END USER CREATED HOOK ################
663 #########################################################
665 return new_user_response
666 except Exception as e:
667 verbose_proxy_logger.exception("/user/new: Exception occured - %s", e)
668 raise handle_exception_on_proxy(e)
671@router.get(
672 "/user/available_roles",
673 tags=["Internal User management"],
674 include_in_schema=False,
675 dependencies=[Depends(user_api_key_auth)],
676)
677async def ui_get_available_role(
678 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
679):
680 """
681 Endpoint used by Admin UI to show all available roles to assign a user
682 return {
683 "proxy_admin": {
684 "description": "Proxy Admin role",
685 "ui_label": "Admin"
686 }
687 }
688 """
690 _data_to_return: Final = {}
691 for role in LitellmUserRoles:
692 # We only show a subset of roles on UI
693 if role in [
694 LitellmUserRoles.PROXY_ADMIN,
695 LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY,
696 LitellmUserRoles.INTERNAL_USER,
697 LitellmUserRoles.INTERNAL_USER_VIEW_ONLY,
698 ]:
699 _data_to_return[role.value] = {
700 "description": role.description,
701 "ui_label": role.ui_label,
702 }
703 return _data_to_return
706def get_team_from_list(
707 team_list: list[LiteLLM_TeamTable] | list[TeamListResponseObject] | None,
708 team_id: str,
709) -> LiteLLM_TeamTable | LiteLLM_TeamMembership | None:
710 if team_list is None:
711 return None
713 for team in team_list:
714 if team.team_id == team_id:
715 return team
716 return None
719def _is_valid_user_id(user_id: str) -> bool:
720 """Validate that a decoded user_id is safe to use downstream."""
721 MAX_USER_ID_LENGTH: Final = 512
722 if len(user_id) > MAX_USER_ID_LENGTH: 722 ↛ 723line 722 didn't jump to line 723 because the condition on line 722 was never true
723 return False
724 # Reject ASCII control characters (U+0000–U+001F)
725 for ch in user_id: 725 ↛ 728line 725 didn't jump to line 728 because the loop on line 725 didn't complete
726 if ord(ch) < 0x20:
727 return False
728 return True
731def get_user_id_from_request(request: Request) -> str | None:
732 """
733 Get the user id from the request
734 """
735 # Get the raw query string and parse it properly to handle + characters
736 user_id: str | None = None
737 query_string: Final = str(request.url.query)
738 if "user_id=" in query_string: 738 ↛ 749line 738 didn't jump to line 749 because the condition on line 738 was always true
739 # Extract the user_id value from the raw query string
740 import re
741 from urllib.parse import unquote
743 match: Final = re.search(r"user_id=([^&]*)", query_string)
744 if match: 744 ↛ 749line 744 didn't jump to line 749 because the condition on line 744 was always true
745 # Use unquote instead of unquote_plus to preserve + characters
746 raw_user_id: Final = unquote(match.group(1))
747 if _is_valid_user_id(raw_user_id): 747 ↛ 748line 747 didn't jump to line 748 because the condition on line 747 was never true
748 user_id = raw_user_id
749 return user_id
752def _normalize_user_info_user_id(request: Request, user_id: str | None) -> str | None:
753 """Normalize URL-decoded user_id while preserving '+' characters."""
754 if user_id is not None and " " in user_id:
755 return get_user_id_from_request(request=request)
756 return user_id
759def _enforce_user_info_access(user_id: str | None, user_api_key_dict: UserAPIKeyAuth) -> None:
760 """Re-validate that the caller may read the resolved ``user_id`` after
761 URL-decoding has been finalized.
763 The route-level check in ``RouteChecks.non_proxy_admin_allowed_routes_check``
764 runs against ``request.query_params``, which decodes a literal ``+`` to a
765 space. ``_normalize_user_info_user_id`` then re-parses the raw query with
766 ``unquote`` so the endpoint can return rows for user_ids that contain ``+``
767 (e.g. plus-addressed emails). That asymmetry let an attacker who registered
768 a username with a literal space pass the route check and then read another
769 user's row by sending the encoded ``+`` form. Re-checking ownership here
770 closes the gap without changing the supported user_id grammar.
771 """
772 if user_id is None:
773 return
774 # Admin-view roles (PROXY_ADMIN and PROXY_ADMIN_VIEW_ONLY) bypass
775 # ownership, mirroring the `/user/info` carve-out that
776 # `RouteChecks.non_proxy_admin_allowed_routes_check` applies upstream.
777 if _user_has_admin_view(user_api_key_dict): 777 ↛ 779line 777 didn't jump to line 779 because the condition on line 777 was always true
778 return
779 if user_id == user_api_key_dict.user_id:
780 return
781 raise HTTPException(
782 status_code=status.HTTP_403_FORBIDDEN,
783 detail=(
784 f"key not allowed to access this user's info. user_id={user_id}, key's user_id={user_api_key_dict.user_id}"
785 ),
786 )
789class _UserInfoDataClient(Protocol):
790 @overload
791 async def get_data(self, *, user_id: str) -> "prisma_models.LiteLLM_UserTable | None": ... 791 ↛ exitline 791 didn't return from function 'get_data' because
793 @overload
794 async def get_data( 794 ↛ exitline 794 didn't return from function 'get_data' because
795 self,
796 *,
797 user_id: str | None,
798 table_name: Literal["key"],
799 query_type: Literal["find_all"],
800 ) -> "Sequence[LiteLLM_VerificationToken] | None": ...
802 @overload
803 async def get_data( 803 ↛ exitline 803 didn't return from function 'get_data' because
804 self,
805 *,
806 team_id_list: list[str],
807 table_name: Literal["team"],
808 query_type: Literal["find_all"],
809 ) -> "Sequence[TeamListResponseObject] | None": ...
812async def _get_user_info_keys(
813 prisma_client: "_UserInfoDataClient",
814 user_id: str | None,
815) -> "Sequence[LiteLLM_VerificationToken] | None":
816 return await prisma_client.get_data(
817 user_id=user_id,
818 table_name="key",
819 query_type="find_all",
820 )
823async def _get_user_info_teams(
824 prisma_client: "_UserInfoDataClient",
825 user_id: str | None,
826 user_info: "prisma_models.LiteLLM_UserTable",
827 user_api_key_dict: UserAPIKeyAuth,
828) -> tuple[list[TeamListResponseObject], list[TeamListResponseObject] | None]:
829 """Fetch and merge teams from membership + user.teams field."""
830 from litellm.proxy.management_endpoints.team_endpoints import list_team
832 team_list: list[TeamListResponseObject] = []
833 team_id_list: list[str] = []
835 teams_1: Final = await list_team(
836 http_request=Request(
837 scope={"type": "http", "path": "/user/info"},
838 ),
839 user_id=user_id,
840 user_api_key_dict=user_api_key_dict,
841 )
843 if teams_1 is not None and isinstance(teams_1, list): 843 ↛ 847line 843 didn't jump to line 847 because the condition on line 843 was always true
844 team_list = teams_1
845 team_id_list = [team.team_id for team in teams_1]
847 teams_2: Sequence[TeamListResponseObject] | None = None
848 target_team_ids: Final = getattr(user_info, "teams", None)
850 if target_team_ids and isinstance(target_team_ids, list): 850 ↛ 856line 850 didn't jump to line 856 because the condition on line 850 was always true
851 teams_2 = await prisma_client.get_data(
852 team_id_list=target_team_ids,
853 table_name="team",
854 query_type="find_all",
855 )
856 elif user_api_key_dict.user_id is not None and user_id is None:
857 caller_user_info: Final = await prisma_client.get_data(user_id=user_api_key_dict.user_id)
858 caller_team_ids: Final = caller_user_info.teams if caller_user_info is not None else None
859 if caller_team_ids:
860 teams_2 = await prisma_client.get_data(
861 team_id_list=caller_team_ids,
862 table_name="team",
863 query_type="find_all",
864 )
866 if teams_2 is not None and isinstance(teams_2, list): 866 ↛ 872line 866 didn't jump to line 872 because the condition on line 866 was always true
867 for team in teams_2:
868 if team.team_id not in team_id_list: 868 ↛ 869line 868 didn't jump to line 869 because the condition on line 868 was never true
869 team_list.append(team)
870 team_id_list.append(team.team_id)
872 return team_list, teams_1
875_SCIM_DIRECTORY_METADATA_KEYS: Final = frozenset(
876 {SCIM_ENTERPRISE_METADATA_KEY, SCIM_ENTITLEMENTS_METADATA_KEY, SCIM_ROLES_METADATA_KEY}
877)
880def _redact_scim_enterprise_metadata(
881 metadata: dict[str, object] | None,
882) -> dict[str, object] | None:
883 """SCIM enterprise attributes, entitlements, and roles are persisted in user
884 metadata so reporting can group on them, but they are directory-only fields
885 that generic user-info endpoints must not surface; SCIM clients read them
886 through the SCIM endpoints."""
887 if not isinstance(metadata, dict) or not _SCIM_DIRECTORY_METADATA_KEYS.intersection(metadata): 887 ↛ 889line 887 didn't jump to line 889 because the condition on line 887 was always true
888 return metadata
889 return {k: v for k, v in metadata.items() if k not in _SCIM_DIRECTORY_METADATA_KEYS}
892def _build_user_info_response(
893 user_id: str | None,
894 user_info: Any | None,
895 keys: Sequence[LiteLLM_VerificationToken] | None,
896 team_list: list[TeamListResponseObject],
897 teams_1: list[TeamListResponseObject] | None,
898 model_max_budget_usage: dict[str, dict[str, object]] | None = None,
899) -> UserInfoResponse:
900 """Create UserInfoResponse while filtering sensitive fields."""
901 if user_info is None and keys is not None: 901 ↛ 902line 901 didn't jump to line 902 because the condition on line 901 was never true
902 spend: Final = sum(getattr(k, "spend", 0) for k in keys)
903 user_info = {"spend": spend}
905 returned_keys: Final = _process_keys_for_user_info(keys=keys, all_teams=teams_1)
906 team_list.sort(key=lambda x: getattr(x, "team_alias", "") or "")
908 _user_info: Final = user_info.model_dump() if isinstance(user_info, BaseModel) else user_info
909 if isinstance(_user_info, dict): 909 ↛ 915line 909 didn't jump to line 915 because the condition on line 909 was always true
910 _user_info.pop("password", None)
911 _user_info["metadata"] = _redact_scim_enterprise_metadata(_user_info.get("metadata"))
912 if model_max_budget_usage is not None: 912 ↛ 915line 912 didn't jump to line 915 because the condition on line 912 was always true
913 _user_info["model_max_budget_usage"] = model_max_budget_usage
915 return UserInfoResponse(
916 user_id=user_id,
917 user_info=_user_info,
918 keys=returned_keys,
919 teams=team_list,
920 )
923@router.get(
924 "/user/info",
925 tags=["Internal User management"],
926 dependencies=[Depends(user_api_key_auth)],
927 response_model=UserInfoResponse,
928)
929@management_endpoint_wrapper
930async def user_info(
931 request: Request,
932 user_id: str | None = fastapi.Query(default=None, description="User ID in the request parameters"),
933 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
934):
935 """
936 [10/07/2024]
937 Note: To get all users (+pagination), use `/user/list` endpoint.
940 Use this to get user information. (user row + all user key info)
942 Example request
943 ```
944 curl -X GET 'http://localhost:4000/user/info?user_id=krrish7%40berri.ai' \
945 --header 'Authorization: Bearer sk-1234'
946 ```
947 """
948 from litellm.proxy.proxy_server import model_max_budget_limiter, prisma_client
950 try:
951 user_id = _normalize_user_info_user_id(request=request, user_id=user_id)
952 _enforce_user_info_access(user_id=user_id, user_api_key_dict=user_api_key_dict)
954 if prisma_client is None: 954 ↛ 955line 954 didn't jump to line 955 because the condition on line 954 was never true
955 raise Exception(
956 "Database not connected. Connect a database to your proxy - https://docs.litellm.ai/docs/simple_proxy#managing-auth---virtual-keys"
957 )
958 if user_id is None and _user_has_admin_view(user_api_key_dict):
959 return await _get_user_info_for_proxy_admin(user_api_key_dict=user_api_key_dict)
960 elif user_id is None: 960 ↛ 961line 960 didn't jump to line 961 because the condition on line 960 was never true
961 user_id = user_api_key_dict.user_id
962 ## GET USER ROW ##
964 user_info = None
965 if user_id is not None: 965 ↛ 968line 965 didn't jump to line 968 because the condition on line 965 was always true
966 user_info = await prisma_client.get_data(user_id=user_id)
968 if user_info is None:
969 raise HTTPException(
970 status_code=404,
971 detail=f"User {user_id} not found",
972 )
974 team_list, teams_1 = await _get_user_info_teams(
975 prisma_client=prisma_client,
976 user_id=user_id,
977 user_info=user_info,
978 user_api_key_dict=user_api_key_dict,
979 )
981 ## GET ALL KEYS ##
982 keys: Final = await _get_user_info_keys(prisma_client, user_id)
984 response_data: Final = _build_user_info_response(
985 user_id=user_id,
986 user_info=user_info,
987 keys=keys,
988 team_list=team_list,
989 teams_1=teams_1,
990 model_max_budget_usage=await build_model_max_budget_usage(
991 entity_type=Litellm_EntityType.USER,
992 entity_id=user_id,
993 model_max_budget=getattr(user_info, "model_max_budget", None),
994 cache=model_max_budget_limiter.dual_cache,
995 ),
996 )
998 return response_data
999 except Exception as e:
1000 verbose_proxy_logger.exception("litellm.proxy.proxy_server.user_info(): Exception occured - %s", e)
1001 raise handle_exception_on_proxy(e)
1004async def _check_user_info_v2_access(
1005 user_api_key_dict: UserAPIKeyAuth,
1006 target_user_id: str,
1007) -> "prisma_models.LiteLLM_UserTable | None":
1008 """
1009 Check if the caller is allowed to access the target user's info.
1011 Returns the target user's DB row if access is allowed, None otherwise.
1012 Returning the row avoids a redundant DB fetch in the caller.
1014 Access rules:
1015 1. Proxy admins / proxy admin viewers can access any user
1016 2. User can access their own info
1017 3. Team admins can access info of users in their teams
1019 Raises on unexpected DB errors so they surface as 500s, not silent 404s.
1020 """
1021 from litellm.proxy.proxy_server import prisma_client
1023 if prisma_client is None: 1023 ↛ 1024line 1023 didn't jump to line 1024 because the condition on line 1023 was never true
1024 return None
1026 # Helper: fetch the target user row (reused across branches). object_permission is included so
1027 # callers can read the user's MCP/vector-store entitlements without a second round trip.
1028 async def _fetch_target_user():
1029 return await _user_table(prisma_client).find_unique(
1030 where={"user_id": target_user_id}, include={"object_permission": True}
1031 )
1033 # Rule 1: Proxy admins — fetch and return the target row directly
1034 if _user_has_admin_view(user_api_key_dict): 1034 ↛ 1038line 1034 didn't jump to line 1038 because the condition on line 1034 was always true
1035 return await _fetch_target_user()
1037 # Rule 2: Self-lookup
1038 if user_api_key_dict.user_id == target_user_id:
1039 return await _fetch_target_user()
1041 # Rule 3: Team admins can look up users in their teams
1042 if user_api_key_dict.user_id is not None:
1043 # Get caller's teams
1044 caller_user: Final = await _user_table(prisma_client).find_unique(where={"user_id": user_api_key_dict.user_id})
1045 if caller_user is not None and caller_user.teams:
1046 # Fetch the target user ONCE, before the loop
1047 target_user: Final = await _fetch_target_user()
1048 if target_user is None:
1049 return None
1051 # Get all teams the caller belongs to
1052 teams: Final = await _team_table(prisma_client).find_many(where={"team_id": {"in": caller_user.teams}})
1053 for team in teams:
1054 team_obj = LiteLLM_TeamTable.model_validate(team.model_dump())
1055 if _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj):
1056 # Check if target user is in this team
1057 if team.team_id in (target_user.teams or []):
1058 return target_user
1060 return None
1063@router.get(
1064 "/v2/user/info",
1065 tags=["Internal User management"],
1066 dependencies=[Depends(user_api_key_auth)],
1067 response_model=UserInfoV2Response,
1068)
1069@management_endpoint_wrapper
1070async def user_info_v2(
1071 request: Request,
1072 user_id: str | None = fastapi.Query(default=None, description="User ID in the request parameters"),
1073 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
1074):
1075 """
1076 Lightweight endpoint to get user info. Returns only the user object — no keys, no teams objects.
1078 This is the v2 replacement for /user/info, designed to avoid the "god endpoint" problem
1079 where the old endpoint loaded all keys and teams into memory.
1081 Note on `spend`: this is the user's running budget counter, which the budget reset job
1082 resets whenever `budget_reset_at` elapses (see `budget_duration`): to zero by default,
1083 or to the overage above `max_budget` when `budget_rollover` is enabled. It is NOT
1084 lifetime or per-period historical spend. For historical spend over a date range, use
1085 `/user/daily/activity` or `/user/daily/activity/aggregated`, which read daily spend
1086 records that only ever accumulate and are never reset. The two values are expected to
1087 diverge once a budget reset has occurred within the queried period.
1089 Access control:
1090 - Proxy admins can query any user
1091 - Team admins can query users within their teams
1092 - Internal users can only query themselves (omit user_id or pass own)
1093 - Returns 404 for non-existent users or unauthorized access
1095 Example request:
1096 ```
1097 curl -X GET 'http://localhost:4000/v2/user/info?user_id=user123' \\
1098 --header 'Authorization: Bearer sk-1234'
1099 ```
1100 """
1101 from litellm.proxy.proxy_server import model_max_budget_limiter, prisma_client
1103 try:
1104 if prisma_client is None: 1104 ↛ 1105line 1104 didn't jump to line 1105 because the condition on line 1104 was never true
1105 raise HTTPException(
1106 status_code=500,
1107 detail=CommonProxyErrors.db_not_connected_error.value,
1108 )
1110 # Handle URL encoding for + characters
1111 if user_id is not None and " " in user_id:
1112 user_id = get_user_id_from_request(request=request)
1114 # Default to self-lookup if no user_id provided
1115 if user_id is None:
1116 user_id = user_api_key_dict.user_id
1118 if user_id is None: 1118 ↛ 1119line 1118 didn't jump to line 1119 because the condition on line 1118 was never true
1119 raise HTTPException(
1120 status_code=400,
1121 detail="user_id is required. Either pass it as a query parameter or authenticate with a user-bound key.",
1122 )
1124 # Check access — returns the user row if allowed, None otherwise.
1125 # This avoids a redundant DB fetch since the access check already
1126 # loads the target user for team-admin verification.
1127 user_row: Final = await _check_user_info_v2_access(
1128 user_api_key_dict=user_api_key_dict,
1129 target_user_id=user_id,
1130 )
1132 if user_row is None:
1133 raise HTTPException(
1134 status_code=404,
1135 detail=f"User not found: {user_id}",
1136 )
1138 user_data: Final = user_row.model_dump()
1140 return UserInfoV2Response(
1141 user_id=user_data.get("user_id", user_id),
1142 user_email=user_data.get("user_email"),
1143 user_alias=user_data.get("user_alias"),
1144 user_role=user_data.get("user_role"),
1145 spend=user_data.get("spend", 0.0),
1146 max_budget=user_data.get("max_budget"),
1147 models=user_data.get("models") or [],
1148 budget_duration=user_data.get("budget_duration"),
1149 budget_reset_at=user_data.get("budget_reset_at"),
1150 metadata=_redact_scim_enterprise_metadata(user_data.get("metadata")),
1151 created_at=user_data.get("created_at"),
1152 updated_at=user_data.get("updated_at"),
1153 sso_user_id=user_data.get("sso_user_id"),
1154 teams=user_data.get("teams") or [],
1155 object_permission=user_data.get("object_permission"),
1156 model_max_budget=user_data.get("model_max_budget"),
1157 model_max_budget_usage=await build_model_max_budget_usage(
1158 entity_type=Litellm_EntityType.USER,
1159 entity_id=user_data.get("user_id", user_id),
1160 model_max_budget=user_data.get("model_max_budget"),
1161 cache=model_max_budget_limiter.dual_cache,
1162 ),
1163 )
1164 except Exception as e:
1165 verbose_proxy_logger.exception("litellm.proxy.proxy_server.user_info_v2(): Exception occured - %s", e)
1166 raise handle_exception_on_proxy(e)
1169async def _fetch_admin_teams_and_keys_rows(
1170 prisma_client: "PrismaClient", sql_query: str
1171) -> Sequence[Mapping[str, Sequence[Mapping[str, object]] | None]]:
1172 return await prisma_client.db.query_raw(sql_query)
1175async def _get_user_info_for_proxy_admin(user_api_key_dict: UserAPIKeyAuth):
1176 """
1177 Admin UI Endpoint - Returns All Teams and Keys when Proxy Admin is querying
1179 - get all teams in LiteLLM_TeamTable
1180 - get all keys in LiteLLM_VerificationToken table
1182 Why separate helper for proxy admin ?
1183 - To get Faster UI load times, get all teams and virtual keys in 1 query
1184 """
1186 from litellm.proxy.proxy_server import prisma_client
1188 sql_query: Final = """
1189 SELECT
1190 (SELECT json_agg(t.*) FROM "LiteLLM_TeamTable" t) as teams,
1191 (SELECT json_agg(k.*) FROM "LiteLLM_VerificationToken" k WHERE k.team_id != 'litellm-dashboard' OR k.team_id IS NULL) as keys
1192 """
1193 if prisma_client is None: 1193 ↛ 1194line 1193 didn't jump to line 1194 because the condition on line 1193 was never true
1194 raise Exception(
1195 "Database not connected. Connect a database to your proxy - https://docs.litellm.ai/docs/simple_proxy#managing-auth---virtual-keys"
1196 )
1198 results: Final = await _fetch_admin_teams_and_keys_rows(prisma_client, sql_query)
1200 verbose_proxy_logger.debug("results_keys: %s", results)
1202 _keys_in_db: Final[Sequence[Mapping[str, object]]] = results[0]["keys"] or []
1203 # cast all keys to LiteLLM_VerificationToken
1204 keys_in_db: Final = []
1205 for key in _keys_in_db:
1206 key_payload = dict[str, object](key)
1207 if key_payload.get("models") is None: 1207 ↛ 1208line 1207 didn't jump to line 1208 because the condition on line 1207 was never true
1208 key_payload["models"] = []
1209 keys_in_db.append(LiteLLM_VerificationToken.model_validate(key_payload))
1211 # cast all teams to LiteLLM_TeamTable
1212 _teams_rows: Final[Sequence[Mapping[str, object]]] = results[0]["teams"] or []
1213 _teams_in_db: Final = sorted(
1214 (LiteLLM_TeamTable.model_validate(team) for team in _teams_rows),
1215 key=lambda x: getattr(x, "team_alias", "") or "",
1216 )
1217 returned_keys: Final = _process_keys_for_user_info(keys=keys_in_db, all_teams=_teams_in_db)
1219 # Get admin's own user_id and user_info
1220 admin_user_id: Final = user_api_key_dict.user_id
1221 admin_user_info = None
1223 if admin_user_id is not None: 1223 ↛ 1232line 1223 didn't jump to line 1232 because the condition on line 1223 was always true
1224 admin_user_info = await prisma_client.get_data(user_id=admin_user_id)
1225 if admin_user_info is not None: 1225 ↛ 1226line 1225 didn't jump to line 1226 because the condition on line 1225 was never true
1226 admin_user_info = (
1227 admin_user_info.model_dump() if isinstance(admin_user_info, BaseModel) else admin_user_info
1228 )
1229 if isinstance(admin_user_info, dict):
1230 admin_user_info.pop("password", None)
1232 return UserInfoResponse(
1233 user_id=admin_user_id,
1234 user_info=admin_user_info,
1235 keys=returned_keys,
1236 teams=_teams_in_db,
1237 )
1240def _process_keys_for_user_info(
1241 keys: Sequence[LiteLLM_VerificationToken] | None,
1242 all_teams: list[LiteLLM_TeamTable] | list[TeamListResponseObject] | None,
1243):
1244 from litellm.constants import UI_SESSION_TOKEN_TEAM_ID
1245 from litellm.proxy.proxy_server import general_settings, litellm_master_key_hash
1247 returned_keys: Final = []
1248 if keys is None: 1248 ↛ 1249line 1248 didn't jump to line 1249 because the condition on line 1248 was never true
1249 pass
1250 else:
1251 for key in keys: 1251 ↛ 1252line 1251 didn't jump to line 1252 because the loop on line 1251 never started
1252 if (
1253 key.token == litellm_master_key_hash
1254 and general_settings.get("disable_master_key_return", False)
1255 is True ## [IMPORTANT] used by hosted proxy-ui to prevent sharing master key on ui
1256 ):
1257 continue
1259 try:
1260 _key: dict = key.model_dump()
1261 except Exception:
1262 # if using pydantic v1
1263 _key = key.dict()
1265 # Filter out UI session tokens (team_id="litellm-dashboard")
1266 if _key.get("team_id") == UI_SESSION_TOKEN_TEAM_ID:
1267 continue
1269 if "team_id" in _key and _key["team_id"] is not None and _key["team_id"] != "litellm-dashboard":
1270 team_info = get_team_from_list(team_list=all_teams, team_id=_key["team_id"])
1271 if team_info is not None:
1272 team_alias = getattr(team_info, "team_alias", None)
1273 _key["team_alias"] = team_alias
1274 else:
1275 _key["team_alias"] = None
1276 else:
1277 _key["team_alias"] = "None"
1278 returned_keys.append(_key)
1279 return returned_keys
1282def _update_internal_user_params(data_json: dict, data: UpdateUserRequest | UpdateUserRequestNoUserIDorEmail) -> dict:
1283 non_default_values: Final = {}
1284 fields_set: Final = data.fields_set() if hasattr(data, "fields_set") else set()
1286 for k, v in data_json.items():
1287 if k in ("max_budget", "budget_duration"):
1288 if k in fields_set: 1288 ↛ 1286line 1288 didn't jump to line 1286 because the condition on line 1288 was always true
1289 non_default_values[k] = v
1290 elif k == "model_max_budget": 1290 ↛ 1291line 1290 didn't jump to line 1291 because the condition on line 1290 was never true
1291 if k in fields_set:
1292 try:
1293 _USER_MODEL_BUDGET_ADAPTER.validate_python({} if v is None else v)
1294 except ValidationError as exc:
1295 raise HTTPException(status_code=400, detail=str(exc)) from exc
1296 non_default_values[k] = {} if v is None else v
1297 elif (
1298 v is not None
1299 and v
1300 not in (
1301 [],
1302 {},
1303 )
1304 and k not in LiteLLM_ManagementEndpoint_MetadataFields
1305 ): # models default to [], spend defaults to 0, we should not reset these values
1306 non_default_values[k] = v
1308 is_internal_user = False
1309 if data.user_role == LitellmUserRoles.INTERNAL_USER: 1309 ↛ 1310line 1309 didn't jump to line 1310 because the condition on line 1309 was never true
1310 is_internal_user = True
1312 if "budget_duration" in non_default_values: 1312 ↛ 1322line 1312 didn't jump to line 1322 because the condition on line 1312 was always true
1313 from litellm.proxy.common_utils.timezone_utils import get_budget_reset_time
1315 validate_budget_duration(non_default_values["budget_duration"])
1316 non_default_values["budget_reset_at"] = (
1317 get_budget_reset_time(budget_duration=non_default_values["budget_duration"])
1318 if non_default_values["budget_duration"] is not None
1319 else None
1320 )
1322 if "max_budget" not in non_default_values:
1323 if (
1324 is_internal_user and litellm.max_internal_user_budget is not None
1325 ): # applies internal user limits, if user role updated
1326 non_default_values["max_budget"] = litellm.max_internal_user_budget
1328 if "budget_duration" not in non_default_values: # applies internal user limits, if user role updated
1329 if is_internal_user and litellm.internal_user_budget_duration is not None:
1330 non_default_values["budget_duration"] = litellm.internal_user_budget_duration
1331 from litellm.proxy.common_utils.timezone_utils import get_budget_reset_time
1333 non_default_values["budget_reset_at"] = get_budget_reset_time(
1334 budget_duration=non_default_values["budget_duration"]
1335 )
1337 return non_default_values
1340async def _schedule_user_update_audit_log(
1341 response: Mapping[str, object],
1342 existing_user_row: BaseModel | None,
1343 litellm_changed_by: str | None,
1344 user_api_key_dict: UserAPIKeyAuth,
1345 litellm_proxy_admin_name: str | None,
1346) -> None:
1347 from litellm.proxy.proxy_server import prisma_client
1349 if prisma_client is None:
1350 return
1351 try:
1352 updated_user_row: Final = await _user_table(prisma_client).find_first(where={"user_id": response["user_id"]})
1353 if updated_user_row:
1354 user_row_typed: Final = LiteLLM_UserTable.model_validate(updated_user_row.model_dump(exclude_none=True))
1355 asyncio.create_task(
1356 UserManagementEventHooks.create_internal_user_audit_log(
1357 user_id=user_row_typed.user_id,
1358 action="updated",
1359 litellm_changed_by=litellm_changed_by or user_api_key_dict.user_id,
1360 user_api_key_dict=user_api_key_dict,
1361 litellm_proxy_admin_name=litellm_proxy_admin_name,
1362 before_value=(existing_user_row.model_dump_json(exclude_none=True) if existing_user_row else None),
1363 after_value=user_row_typed.model_dump_json(exclude_none=True),
1364 )
1365 )
1366 except Exception as audit_error:
1367 verbose_proxy_logger.warning("Failed to create audit log for user %s: %s", response.get("user_id"), audit_error)
1370def _check_user_update_authz(
1371 user_request: UpdateUserRequest,
1372 user_api_key_dict: UserAPIKeyAuth,
1373 existing_user_row: BaseModel | None,
1374) -> None:
1375 """Authorization checks for /user/update — raises HTTPException on failure."""
1376 if user_request.user_role is not None and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value:
1377 raise HTTPException(status_code=403, detail="Only proxy admins can modify user roles.")
1379 if existing_user_row is not None:
1380 typed_row: Final = LiteLLM_UserTable.model_validate(existing_user_row.model_dump(exclude_none=True))
1381 if not can_user_call_user_update(user_api_key_dict=user_api_key_dict, user_info=typed_row):
1382 raise HTTPException(
1383 status_code=403,
1384 detail={
1385 "error": "User does not have permission to update this user. Only PROXY_ADMIN can update other users."
1386 },
1387 )
1388 elif user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value:
1389 # Silent-create guard: only PROXY_ADMIN may create via /user/update.
1390 raise HTTPException(
1391 status_code=404,
1392 detail={
1393 "error": "User not found. Only PROXY_ADMIN can create users via /user/update; use /user/new instead."
1394 },
1395 )
1398async def _invalidate_user_spend_counter_if_changed(
1399 non_default_values: Mapping[str, object],
1400) -> None:
1401 """Invalidate the cross-pod spend counter after a direct ``spend`` change.
1403 A direct ``spend`` change must also invalidate the cross-pod spend counter
1404 enforcement reads; the DB write alone leaves a warm counter at the stale
1405 value. ``non_default_values["user_id"]`` is populated in every branch of the
1406 caller (incl. the email-new-user insert path, whose response is a bare model
1407 and not safely subscriptable).
1408 """
1409 if non_default_values.get("spend") is not None:
1410 from litellm.proxy.proxy_server import _invalidate_spend_counter
1412 await _invalidate_spend_counter(counter_key=f"spend:user:{non_default_values['user_id']}")
1415def _clears_object_permission(user_request: UpdateUserRequest) -> bool:
1416 """Whether the caller explicitly asked to remove this user's object_permission.
1418 Distinguishes "sent nothing" from "sent an empty grant set". Only the latter clears; an omitted
1419 field must leave an existing entitlement alone.
1420 """
1421 if "object_permission" not in (user_request.fields_set() if hasattr(user_request, "fields_set") else set()):
1422 return False
1423 sent: Final = user_request.object_permission
1424 return sent is None or not sent.model_dump(exclude_unset=True, exclude_none=True)
1427async def _invalidate_cached_user_entitlement(user_id: str | None, object_permission_ids: tuple[str, ...]) -> None:
1428 """Drop the cache entries an entitlement change makes stale.
1430 All three kinds are needed: a permission row is cached under its own id (so re-reading the same
1431 link still yields the OLD grants), the ``user_id -> object_permission_id`` link is cached
1432 separately (so a user who previously had NO entitlement keeps its "none" sentinel), and the user
1433 row itself is cached whole. Leaving any behind means an admin revoking a tool keeps serving it
1434 until the management-object TTL expires.
1436 Both the outgoing and incoming permission ids are passed, because a clear leaves no incoming id
1437 at all and an upsert may mint a new row; invalidating only one of the two leaves the other's
1438 grants live.
1440 Each deletion is isolated: one that fails must not skip the others, or a single unreachable key
1441 would silently leave the rest of a revocation in place. Best-effort overall, exactly as the caches
1442 are everywhere else, since one we cannot clear still expires on its own.
1443 """
1444 from litellm.proxy.proxy_server import user_api_key_cache
1446 keys: Final = (
1447 *(object_permission_cache_key(permission_id) for permission_id in dict.fromkeys(object_permission_ids)),
1448 *((user_object_permission_id_cache_key(user_id), user_id) if user_id is not None else ()),
1449 )
1450 for key in keys:
1451 try:
1452 await user_api_key_cache.async_delete_cache(key=key)
1453 except Exception as e: # noqa: BLE001 # a cache we cannot clear still expires; never fail the write
1454 verbose_proxy_logger.warning("Failed to invalidate cached entitlement key %r: %s", key, e)
1457async def _update_single_user_helper(
1458 user_request: UpdateUserRequest,
1459 user_api_key_dict: UserAPIKeyAuth,
1460 litellm_changed_by: str | None = None,
1461 password_prevalidated: bool = False,
1462) -> dict[str, Any]:
1463 """
1464 Helper function to update a single user.
1465 Used by both user_update and bulk_user_update endpoints.
1467 Returns the updated user data or raises an exception on failure.
1468 """
1469 from litellm.proxy.proxy_server import general_settings, litellm_proxy_admin_name, prisma_client, user_api_key_cache
1471 if prisma_client is None: 1471 ↛ 1472line 1471 didn't jump to line 1472 because the condition on line 1471 was never true
1472 raise Exception("Not connected to DB!")
1474 if not user_request.user_id and not user_request.user_email:
1475 raise ValueError("Either user_id or user_email must be provided")
1477 _check_permissions_caller_permission(
1478 data=user_request,
1479 user_api_key_dict=user_api_key_dict,
1480 )
1482 data_json: Final[dict] = user_request.model_dump(exclude_unset=True)
1483 non_default_values = _update_internal_user_params(data_json=data_json, data=user_request)
1484 await _hash_password_in_dict(non_default_values, general_settings, password_prevalidated=password_prevalidated)
1486 existing_user_row: BaseModel | None = None
1487 if user_request.user_id:
1488 existing_user_row = await _user_table(prisma_client).find_first(where={"user_id": user_request.user_id})
1489 elif user_request.user_email:
1490 existing_user_row = await _user_table(prisma_client).find_first(where={"user_email": user_request.user_email})
1492 _check_user_update_authz(user_request, user_api_key_dict, existing_user_row)
1494 if existing_user_row is not None:
1495 existing_user_row = LiteLLM_UserTable.model_validate(existing_user_row.model_dump(exclude_none=True))
1497 # Prevent budget self-escalation (GHSA-wvg4-6222-3q4r): non-admin callers
1498 # must not be able to raise their own budget/spend fields.
1499 # can_user_call_user_update() already restricts non-admins to self-updates,
1500 # so this guard only fires for self-escalation attempts.
1501 _target_user_id: Final = user_request.user_id or (
1502 getattr(existing_user_row, "user_id", None) if existing_user_row is not None else None
1503 )
1504 _is_self_update: Final = _target_user_id is not None and user_api_key_dict.user_id == _target_user_id
1505 if _is_self_update and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value:
1506 # object_permission is a CEILING on what this human may reach, so a self-write is an
1507 # escalation path: sending an empty grant list means "no restriction" and would lift a
1508 # restriction an admin placed on them. Checked against the fields the caller actually SENT,
1509 # because `_update_internal_user_params` drops empty values, and `object_permission: {}` is
1510 # precisely the clear-my-own-ceiling case this must refuse.
1511 _sent_fields: Final = user_request.fields_set() if hasattr(user_request, "fields_set") else set()
1512 _protected_fields: Final = ("max_budget", "model_max_budget", "soft_budget", "spend", "object_permission")
1513 for _field in _protected_fields:
1514 if _field in non_default_values or _field in _sent_fields:
1515 raise HTTPException(
1516 status_code=403,
1517 detail={
1518 "error": f"Non-admin users cannot modify '{_field}' on their own record. Contact your proxy admin."
1519 },
1520 )
1522 existing_metadata: Final = (
1523 cast(dict, getattr(existing_user_row, "metadata", {}) or {}) if existing_user_row is not None else {}
1524 )
1526 non_default_values = prepare_metadata_fields(
1527 data=user_request,
1528 non_default_values=non_default_values,
1529 existing_metadata=existing_metadata or {},
1530 )
1532 # Reject NaN/±inf spend before it can reach the DB / spend counter.
1533 validate_finite_spend(non_default_values.get("spend"))
1535 # Upsert the grants into their own row and link it, mirroring /key/update and /team/update.
1536 # This also removes object_permission from the payload, which is not a column on the user table.
1537 if "object_permission" in non_default_values:
1538 object_permission_id: Final = await handle_update_object_permission_common(
1539 data_json=non_default_values,
1540 existing_object_permission_id=getattr(existing_user_row, "object_permission_id", None),
1541 prisma_client=prisma_client,
1542 )
1543 if object_permission_id is not None:
1544 non_default_values["object_permission_id"] = object_permission_id
1545 elif _clears_object_permission(user_request):
1546 # An explicit `{}` or null means "no object permission", which the merge-based upsert cannot
1547 # express: merging an empty grant set over the existing row leaves every grant in place. So
1548 # the link is dropped instead, which is what makes the documented clear actually clear.
1549 non_default_values["object_permission_id"] = None
1551 # Perform the update
1552 response: dict[str, Any] | None = None
1554 if user_request.user_id and len(user_request.user_id) > 0:
1555 non_default_values["user_id"] = user_request.user_id
1556 response = await prisma_client.update_data(
1557 user_id=user_request.user_id,
1558 data=non_default_values,
1559 table_name="user",
1560 )
1561 elif user_request.user_email:
1562 # Handle email-based updates
1563 existing_user_rows: Final = await prisma_client.get_data(
1564 key_val={"user_email": user_request.user_email},
1565 table_name="user",
1566 query_type="find_all",
1567 )
1569 if existing_user_rows and isinstance(existing_user_rows, list) and len(existing_user_rows) > 0:
1570 for existing_user in existing_user_rows:
1571 non_default_values["user_id"] = existing_user.user_id
1572 response = await prisma_client.update_data(
1573 user_id=existing_user.user_id,
1574 data=non_default_values,
1575 table_name="user",
1576 )
1577 break # Update first matching user
1578 else:
1579 # Create new user if not found
1580 non_default_values["user_id"] = str(uuid.uuid4())
1581 non_default_values["user_email"] = user_request.user_email
1582 inserted_user_row: Final = await prisma_client.insert_data(data=non_default_values, table_name="user")
1583 response = inserted_user_row # pyright: ignore[reportAssignmentType] # insert_data returns a prisma row
1585 if response is not None:
1586 if "password" in non_default_values:
1587 # An admin set this user's password, which implies the old one may be
1588 # compromised; kill every existing UI session for the target. Revoke-all
1589 # (no keep) — the caller is the admin, not the target, so the caller's
1590 # own session is not among these.
1591 from litellm.proxy.management_endpoints.session_endpoints import (
1592 revoke_ui_session_keys,
1593 )
1595 target_user_id: Final = non_default_values.get("user_id")
1596 if isinstance(target_user_id, str):
1597 await revoke_ui_session_keys(
1598 user_id=target_user_id,
1599 user_api_key_dict=user_api_key_dict,
1600 litellm_changed_by=litellm_changed_by,
1601 )
1603 await _schedule_user_update_audit_log(
1604 response=response,
1605 existing_user_row=existing_user_row,
1606 litellm_changed_by=litellm_changed_by,
1607 user_api_key_dict=user_api_key_dict,
1608 litellm_proxy_admin_name=litellm_proxy_admin_name,
1609 )
1611 await _invalidate_user_spend_counter_if_changed(non_default_values)
1613 if not _USER_BUDGET_CACHE_FIELDS.isdisjoint(non_default_values) or "metadata" in data_json:
1614 await evict_and_broadcast(
1615 cache_keys=(non_default_values["user_id"],),
1616 user_api_key_cache=user_api_key_cache,
1617 )
1619 if "object_permission_id" in non_default_values:
1620 await _invalidate_cached_user_entitlement(
1621 user_id=non_default_values.get("user_id"),
1622 object_permission_ids=tuple(
1623 permission_id
1624 for permission_id in (
1625 getattr(existing_user_row, "object_permission_id", None),
1626 non_default_values.get("object_permission_id"),
1627 )
1628 if isinstance(permission_id, str)
1629 ),
1630 )
1632 if response is None:
1633 raise HTTPException(
1634 status_code=400,
1635 detail={"error": "Failed to update user"},
1636 )
1637 _strip_password_from_response(response)
1638 return response
1641def can_user_call_user_update(
1642 user_api_key_dict: UserAPIKeyAuth,
1643 user_info: LiteLLM_UserTable,
1644) -> bool:
1645 """
1646 Helper to check if the user has access to the key's info
1647 """
1648 if (
1649 user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN.value
1650 or user_api_key_dict.user_id == user_info.user_id
1651 ):
1652 return True
1653 return False
1656@router.post(
1657 "/user/update",
1658 tags=["Internal User management"],
1659 dependencies=[Depends(user_api_key_auth)],
1660)
1661@management_endpoint_wrapper
1662async def user_update(
1663 data: UpdateUserRequest,
1664 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
1665):
1666 """
1667 Example curl
1669 ```
1670 curl --location 'http://0.0.0.0:4000/user/update' \
1671 --header 'Authorization: Bearer sk-1234' \
1672 --header 'Content-Type: application/json' \
1673 --data '{
1674 "user_id": "test-litellm-user-4",
1675 "user_role": "proxy_admin_viewer"
1676 }'
1677 ```
1679 Parameters:
1680 - user_id: Optional[str] - Specify a user id. If not set, a unique id will be generated.
1681 - user_email: Optional[str] - Specify a user email.
1682 - password: Optional[str] - Set the user's password (admin only). Must satisfy the configured password policy. The user is required to change it at their next login. Users change their own password with POST /user/password/change.
1683 - user_alias: Optional[str] - A descriptive name for you to know who this user id refers to.
1684 - teams: Optional[list] - specify a list of team id's a user belongs to.
1685 - send_invite_email: Optional[bool] - Specify if an invite email should be sent.
1686 - user_role: Optional[str] - Specify a user role - "proxy_admin", "proxy_admin_viewer", "internal_user", "internal_user_viewer", "team", "customer". Info about each role here: `https://github.com/BerriAI/litellm/litellm/proxy/_types.py#L20`
1687 - max_budget: Optional[float] - Specify max budget for a given user.
1688 - budget_duration: Optional[str] - Budget is reset at the end of specified duration. If not set, budget is never reset. You can set duration as seconds ("30s"), minutes ("30m"), hours ("30h"), days ("30d"), months ("1mo").
1689 - models: Optional[list] - Model_name's a user is allowed to call. (if empty, key is allowed to call all models)
1690 - tpm_limit: Optional[int] - Specify tpm limit for a given user (Tokens per minute)
1691 - rpm_limit: Optional[int] - Specify rpm limit for a given user (Requests per minute)
1692 - auto_create_key: bool - Default=True. Flag used for returning a key as part of the /user/new response
1693 - aliases: Optional[dict] - Model aliases for the user - [Docs](https://litellm.vercel.app/docs/proxy/virtual_keys#model-aliases)
1694 - config: Optional[dict] - [DEPRECATED PARAM] User-specific config.
1695 - allowed_cache_controls: Optional[list] - List of allowed cache control values. Example - ["no-cache", "no-store"]. See all values - https://docs.litellm.ai/docs/proxy/caching#turn-on--off-caching-per-request-
1696 - blocked: Optional[bool] - [Not Implemented Yet] Whether the user is blocked.
1697 - guardrails: Optional[List[str]] - [Not Implemented Yet] List of active guardrails for the user
1698 - policies: Optional[List[str]] - List of policy names to apply to the user. Policies define guardrails, conditions, and inheritance rules.
1699 - permissions: Optional[dict] - [Not Implemented Yet] User-specific permissions, eg. turning off pii masking.
1700 - metadata: Optional[dict] - Metadata for user, store information for user. Example metadata = {"team": "core-infra", "app": "app2", "email": "ishaan@berri.ai" }
1701 - max_parallel_requests: Optional[int] - Rate limit a user based on the number of parallel requests. Raises 429 error, if user's parallel requests > x.
1702 - model_max_budget: Optional[dict] - Model-specific max budget for user. [Docs](https://docs.litellm.ai/docs/proxy/users#add-model-specific-budgets-to-keys)
1703 - budget_fallbacks: Optional[Dict[str, List[str]]] - Per-model fallback chain tried in order when that model's own `model_max_budget` is exceeded, e.g. {"gpt-4o": ["gpt-4o-mini"]}.
1704 - model_rpm_limit: Optional[float] - Model-specific rpm limit for user. [Docs](https://docs.litellm.ai/docs/proxy/users#add-model-specific-limits-to-keys)
1705 - mcp_rpm_limit: Optional[dict] - Per-MCP-server rpm limit, keyed by MCP server name {"github": 100, "slack": 200}. Enforced for keys and teams only; values set on a user are stored but not enforced per user.
1706 - tag_rpm_limit: Optional[dict] - Per-request-tag rpm limit, keyed by request tag {"cell-1": 1000, "cell-2": 500}. Enforced for keys only; values set on a user are stored but not enforced per user.
1707 - model_tpm_limit: Optional[float] - Model-specific tpm limit for user. [Docs](https://docs.litellm.ai/docs/proxy/users#add-model-specific-limits-to-keys)
1708 - spend: Optional[float] - Amount spent by user. Default is 0. Will be updated by proxy whenever user is used. You can set duration as seconds ("30s"), minutes ("30m"), hours ("30h"), days ("30d"), months ("1mo").
1709 - agent_id: Optional[str] - The agent id associated with the user.
1710 - team_id: Optional[str] - [DEPRECATED PARAM] The team id of the user. Default is None.
1711 - duration: Optional[str] - [NOT IMPLEMENTED].
1712 - key_alias: Optional[str] - [NOT IMPLEMENTED].
1713 - object_permission: Optional[LiteLLM_ObjectPermissionBase] - internal user-specific object permission. Example - {"vector_stores": ["vector_store_1"], "mcp_servers": ["github"], "mcp_tool_permissions": {"github": ["list_issues"]}}. The MCP grants act as a ceiling on every key this user holds. IF null or {} then no object permission.
1714 - prompts: Optional[List[str]] - List of allowed prompts for the user. If specified, the user will only be able to use these specific prompts.
1715 - budget_limits: Optional[list] - List of concurrent budget windows for the user. Each window specifies a budget_limit, time_period, and optional budget_duration. Example - [{"budget_limit": 10.0, "time_period": "1d"}, {"budget_limit": 50.0, "time_period": "7d"}].
1717 """
1718 try:
1719 verbose_proxy_logger.debug("/user/update: Received data = %s", data)
1721 response: Final = await _update_single_user_helper(
1722 user_request=data,
1723 user_api_key_dict=user_api_key_dict,
1724 )
1725 return response
1726 except Exception as e:
1727 verbose_proxy_logger.exception("litellm.proxy.proxy_server.user_update(): Exception occured - %s", e)
1728 verbose_proxy_logger.debug(traceback.format_exc())
1729 if isinstance(e, HTTPException):
1730 raise ProxyException(
1731 message=getattr(e, "detail", f"Authentication Error({e})"),
1732 type=ProxyErrorTypes.auth_error,
1733 param=getattr(e, "param", "None"),
1734 code=getattr(e, "status_code", status.HTTP_400_BAD_REQUEST),
1735 )
1736 elif isinstance(e, ProxyException): 1736 ↛ 1737line 1736 didn't jump to line 1737 because the condition on line 1736 was never true
1737 raise e
1738 raise ProxyException(
1739 message="Authentication Error, " + str(e),
1740 type=ProxyErrorTypes.auth_error,
1741 param=getattr(e, "param", "None"),
1742 code=status.HTTP_400_BAD_REQUEST,
1743 )
1746async def bulk_update_processed_users(
1747 users_to_update: list[UpdateUserRequest],
1748 user_api_key_dict: UserAPIKeyAuth,
1749 litellm_changed_by: str | None = None,
1750 hibp_client: AsyncHTTPHandler | None = None,
1751) -> BulkUpdateUserResponse:
1752 from litellm.proxy.proxy_server import general_settings
1754 results: Final[list[UserUpdateResult]] = []
1755 successful_updates = 0
1756 failed_updates = 0
1758 # Screen the batch's passwords upfront and concurrently: done per-user
1759 # inside the loop below, each HIBP lookup would be awaited serially and a
1760 # degraded-slow HIBP could stretch a full batch to minutes, timing out the
1761 # request after some updates already persisted.
1762 password_verdicts: Final = await validate_passwords_bulk(
1763 tuple(u.password for u in users_to_update if u.password is not None),
1764 general_settings,
1765 client=hibp_client,
1766 )
1768 # Process each user update independently
1769 try:
1770 for user_request in users_to_update:
1771 try:
1772 if (
1773 user_request.password is not None
1774 and (password_error := password_verdicts.get(user_request.password)) is not None
1775 ):
1776 raise password_error
1777 response = await _update_single_user_helper(
1778 user_request=user_request,
1779 user_api_key_dict=user_api_key_dict,
1780 litellm_changed_by=litellm_changed_by,
1781 password_prevalidated=True,
1782 )
1783 # Record success
1784 results.append(
1785 UserUpdateResult(
1786 user_id=(response.get("user_id") if response else user_request.user_id),
1787 user_email=user_request.user_email,
1788 success=True,
1789 updated_user=response,
1790 )
1791 )
1792 successful_updates += 1
1793 except Exception as e:
1794 verbose_proxy_logger.exception(
1795 "Failed to update user %s: %s", user_request.user_id or user_request.user_email, e
1796 )
1797 # Record failure
1798 error_message = str(e)
1799 verbose_proxy_logger.error(
1800 "Failed to update user %s: %s", user_request.user_id or user_request.user_email, error_message
1801 )
1803 results.append(
1804 UserUpdateResult(
1805 user_id=user_request.user_id,
1806 user_email=user_request.user_email,
1807 success=False,
1808 error=error_message,
1809 )
1810 )
1811 failed_updates += 1
1813 return BulkUpdateUserResponse(
1814 results=results,
1815 total_requested=len(users_to_update),
1816 successful_updates=successful_updates,
1817 failed_updates=failed_updates,
1818 )
1819 except Exception as e:
1820 verbose_proxy_logger.exception("Failed to update users: %s", e)
1821 raise HTTPException(status_code=500, detail={"error": str(e)})
1824@router.post(
1825 "/user/bulk_update",
1826 tags=["Internal User management"],
1827 dependencies=[Depends(user_api_key_auth)],
1828 response_model=BulkUpdateUserResponse,
1829)
1830@management_endpoint_wrapper
1831async def bulk_user_update(
1832 data: BulkUpdateUserRequest,
1833 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
1834 litellm_changed_by: str | None = Header(
1835 None,
1836 description="The litellm-changed-by header enables tracking of actions performed by authorized users on behalf of other users, providing an audit trail for accountability",
1837 ),
1838):
1839 """
1840 Bulk update multiple users at once.
1842 This endpoint allows updating multiple users in a single request. Each user update
1843 is processed independently - if some updates fail, others will still succeed.
1845 Parameters:
1846 - users: Optional[List[UpdateUserRequest]] - List of specific user update requests
1847 - all_users: Optional[bool] - Set to true to update all users in the system
1848 - user_updates: Optional[UpdateUserRequest] - Updates to apply when all_users=True
1850 Returns:
1851 - results: List of individual update results
1852 - total_requested: Total number of users requested for update
1853 - successful_updates: Number of successful updates
1854 - failed_updates: Number of failed updates
1856 Example request for specific users:
1857 ```bash
1858 curl --location 'http://0.0.0.0:4000/user/bulk_update' \
1859 --header 'Authorization: Bearer sk-1234' \
1860 --header 'Content-Type: application/json' \
1861 --data '{
1862 "users": [
1863 {
1864 "user_id": "user1",
1865 "user_role": "internal_user",
1866 "max_budget": 100.0
1867 },
1868 {
1869 "user_email": "user2@example.com",
1870 "user_role": "internal_user_viewer",
1871 "max_budget": 50.0
1872 }
1873 ]
1874 }'
1875 ```
1877 Example request for all users:
1878 ```bash
1879 curl --location 'http://0.0.0.0:4000/user/bulk_update' \
1880 --header 'Authorization: Bearer sk-1234' \
1881 --header 'Content-Type: application/json' \
1882 --data '{
1883 "all_users": true,
1884 "user_updates": {
1885 "user_role": "internal_user",
1886 "max_budget": 50.0
1887 }
1888 }'
1889 ```
1890 """
1891 from litellm.proxy.proxy_server import litellm_proxy_admin_name, prisma_client, user_api_key_cache
1893 if prisma_client is None: 1893 ↛ 1894line 1893 didn't jump to line 1894 because the condition on line 1893 was never true
1894 raise HTTPException(
1895 status_code=500,
1896 detail={"error": "Database not connected"},
1897 )
1899 # Only proxy admins can modify user_role in bulk updates
1900 _bulk_role = getattr(data.user_updates, "user_role", None) if data.user_updates else None
1901 if _bulk_role is None and data.users: 1901 ↛ 1902line 1901 didn't jump to line 1902 because the condition on line 1901 was never true
1902 _bulk_role = next((u.user_role for u in data.users if u.user_role is not None), None)
1903 if _bulk_role is not None and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value: 1903 ↛ 1904line 1903 didn't jump to line 1904 because the condition on line 1903 was never true
1904 raise HTTPException(
1905 status_code=403,
1906 detail="Only proxy admins can modify user roles.",
1907 )
1909 # Determine the list of users to update
1910 users_to_update: list[UpdateUserRequest] | list[UpdateUserRequestNoUserIDorEmail] = []
1912 if data.all_users and data.user_updates: 1912 ↛ 1914line 1912 didn't jump to line 1914 because the condition on line 1912 was never true
1913 # Only proxy admins can update all users at once
1914 if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value:
1915 raise HTTPException(
1916 status_code=403,
1917 detail="Only proxy admins can update all users at once.",
1918 )
1919 if data.user_updates.password is not None:
1920 bulk_password_error: Final[HTTPExceptionErrorDetail] = {
1921 "error": (
1922 "Setting one password for all users is not supported. "
1923 "Use per-user updates via the 'users' list instead."
1924 )
1925 }
1926 raise HTTPException(status_code=400, detail=bulk_password_error)
1927 # Optimized path for updating all users directly in database
1928 all_users_in_db: Final = await _user_table(prisma_client).find_many(order={"created_at": "desc"})
1930 if not all_users_in_db:
1931 raise HTTPException(
1932 status_code=400,
1933 detail={"error": "No users found to update"},
1934 )
1936 # Limit batch size to prevent overwhelming the system
1937 MAX_BATCH_SIZE = 500 # Increased limit for all-users operations
1938 if len(all_users_in_db) > MAX_BATCH_SIZE:
1939 raise HTTPException(
1940 status_code=400,
1941 detail={
1942 "error": f"Maximum {MAX_BATCH_SIZE} users can be updated at once. Found {len(all_users_in_db)} users."
1943 },
1944 )
1946 # Apply update transformations (reuse existing logic)
1947 data_json: Final[dict] = data.user_updates.model_dump(exclude_unset=True)
1948 non_default_values: Final[dict[str, object]] = _update_internal_user_params(
1949 data_json=data_json, data=data.user_updates
1950 )
1952 # Remove user identification fields since we're updating by user_id
1953 non_default_values.pop("user_id", None)
1954 non_default_values.pop("user_email", None)
1956 successful_updates = 0
1957 failed_updates: Final = 0
1958 results: Final[list[UserUpdateResult]] = []
1960 try:
1961 # Perform bulk database update
1962 await UserRepository(prisma_client).table.update_many(
1963 where={},
1964 data=(
1965 {**non_default_values, "model_max_budget": json.dumps(non_default_values["model_max_budget"])}
1966 if "model_max_budget" in non_default_values
1967 else non_default_values
1968 ),
1969 )
1971 if not _USER_BUDGET_CACHE_FIELDS.isdisjoint(non_default_values):
1972 for start in range(0, len(all_users_in_db), _USER_BUDGET_CACHE_INVALIDATION_BATCH_SIZE):
1973 await asyncio.gather(
1974 *(
1975 evict_and_broadcast(cache_keys=(user.user_id,), user_api_key_cache=user_api_key_cache)
1976 for user in all_users_in_db[start : start + _USER_BUDGET_CACHE_INVALIDATION_BATCH_SIZE]
1977 )
1978 )
1980 # Create individual success results
1981 for user in all_users_in_db:
1982 results.append(
1983 UserUpdateResult(
1984 user_id=user.user_id,
1985 user_email=user.user_email,
1986 success=True,
1987 updated_user={"user_id": user.user_id, **non_default_values},
1988 )
1989 )
1990 successful_updates += 1
1992 # Create single audit log entry for bulk operation
1993 try:
1994 asyncio.create_task(
1995 UserManagementEventHooks.create_internal_user_audit_log(
1996 user_id=user_api_key_dict.user_id or "",
1997 action="updated",
1998 litellm_changed_by=litellm_changed_by or user_api_key_dict.user_id,
1999 user_api_key_dict=user_api_key_dict,
2000 litellm_proxy_admin_name=litellm_proxy_admin_name,
2001 before_value=f"Updated {len(all_users_in_db)} users",
2002 after_value=json.dumps(non_default_values),
2003 )
2004 )
2005 except Exception as audit_error:
2006 verbose_proxy_logger.warning("Failed to create bulk audit log: %s", audit_error)
2008 except Exception as e:
2009 verbose_proxy_logger.exception("Failed to perform bulk update: %s", e)
2010 # Fall back to individual updates if bulk update fails
2011 for user in all_users_in_db:
2012 user_update_request = data.user_updates.model_copy()
2013 user_update_request.user_id = user.user_id
2014 users_to_update.append(user_update_request)
2016 if successful_updates > 0:
2017 return BulkUpdateUserResponse(
2018 results=results,
2019 total_requested=len(all_users_in_db),
2020 successful_updates=successful_updates,
2021 failed_updates=failed_updates,
2022 )
2024 elif data.users: 2024 ↛ 2025line 2024 didn't jump to line 2025 because the condition on line 2024 was never true
2025 users_to_update = data.users
2026 else:
2027 raise HTTPException(
2028 status_code=400,
2029 detail={
2030 "error": "Must specify either 'users' for individual updates or 'all_users=True' with 'user_updates' for bulk updates"
2031 },
2032 )
2034 if not users_to_update:
2035 raise HTTPException(
2036 status_code=400,
2037 detail={"error": "No users found to update"},
2038 )
2040 # Limit batch size to prevent overwhelming the system
2041 MAX_BATCH_SIZE = 500 # Increased limit for all-users operations
2042 if len(users_to_update) > MAX_BATCH_SIZE:
2043 raise HTTPException(
2044 status_code=400,
2045 detail={
2046 "error": f"Maximum {MAX_BATCH_SIZE} users can be updated at once. Found {len(users_to_update)} users."
2047 },
2048 )
2050 return await bulk_update_processed_users(
2051 users_to_update=cast(list[UpdateUserRequest], users_to_update),
2052 user_api_key_dict=user_api_key_dict,
2053 litellm_changed_by=litellm_changed_by,
2054 )
2057async def get_user_key_counts(
2058 prisma_client: "PrismaClient | None",
2059 user_ids: list[str] | None = None,
2060) -> Mapping[str, int]:
2061 """
2062 Helper function to get the count of keys for each user using Prisma's count method.
2064 Args:
2065 prisma_client: The Prisma client instance
2066 user_ids: List of user IDs to get key counts for
2068 Returns:
2069 Dictionary mapping user_id to key count
2070 """
2071 from litellm.constants import UI_SESSION_TOKEN_TEAM_ID
2073 if not user_ids or len(user_ids) == 0:
2074 return {}
2076 result: Final[dict[str, int]] = {}
2078 # Get count for each user_id individually
2079 for user_id in user_ids:
2080 count = await _verification_token_table(prisma_client).count(
2081 where={
2082 "user_id": user_id,
2083 "OR": [
2084 {"team_id": None},
2085 {"team_id": {"not": UI_SESSION_TOKEN_TEAM_ID}},
2086 ],
2087 }
2088 )
2089 result[user_id] = count
2091 return result
2094def _validate_sort_params(sort_by: str | None, sort_order: str) -> dict[str, str] | None:
2095 order_by: Final[dict[str, str]] = {}
2097 if sort_by is None: 2097 ↛ 2098line 2097 didn't jump to line 2098 because the condition on line 2097 was never true
2098 return None
2099 # Validate sort_by is a valid column
2100 valid_columns: Final = [
2101 "user_id",
2102 "user_email",
2103 "created_at",
2104 "spend",
2105 "user_alias",
2106 "user_role",
2107 ]
2108 if sort_by not in valid_columns: 2108 ↛ 2115line 2108 didn't jump to line 2115 because the condition on line 2108 was always true
2109 raise HTTPException(
2110 status_code=400,
2111 detail={"error": f"Invalid sort column. Must be one of: {', '.join(valid_columns)}"},
2112 )
2114 # Validate sort_order
2115 if sort_order.lower() not in ["asc", "desc"]:
2116 raise HTTPException(
2117 status_code=400,
2118 detail={"error": "Invalid sort order. Must be 'asc' or 'desc'"},
2119 )
2121 order_by[sort_by] = sort_order.lower()
2123 return order_by
2126async def _authorize_user_list_request(
2127 user_api_key_dict: UserAPIKeyAuth,
2128 organization_ids: str | None,
2129 prisma_client: "PrismaClient | None",
2130 user_api_key_cache: "UserApiKeyCache",
2131 proxy_logging_obj: "ProxyLogging | None",
2132) -> str | None:
2133 """
2134 Authorize the /user/list request and return the (possibly scoped) organization_ids string.
2136 - Proxy admins: returns organization_ids unchanged (may be None).
2137 - Org admins: returns comma-separated org IDs scoped to their allowed orgs.
2138 - Others: raises 403.
2139 """
2140 if _user_has_admin_view(user_api_key_dict): 2140 ↛ 2143line 2140 didn't jump to line 2143 because the condition on line 2140 was always true
2141 return organization_ids
2143 if user_api_key_dict.user_id is None:
2144 raise HTTPException(
2145 status_code=403,
2146 detail={"error": "Only proxy admins and organization admins can list users."},
2147 )
2148 try:
2149 caller_user: Final = await get_user_object(
2150 user_id=user_api_key_dict.user_id,
2151 prisma_client=prisma_client,
2152 user_api_key_cache=user_api_key_cache,
2153 user_id_upsert=False,
2154 proxy_logging_obj=proxy_logging_obj,
2155 )
2156 except ValueError:
2157 raise HTTPException(
2158 status_code=403,
2159 detail={"error": "Only proxy admins and organization admins can list users."},
2160 )
2161 if caller_user is None:
2162 raise HTTPException(
2163 status_code=403,
2164 detail={"error": "Only proxy admins and organization admins can list users."},
2165 )
2167 allowed_org_ids = [
2168 m.organization_id
2169 for m in (caller_user.organization_memberships or [])
2170 if m.user_role == LitellmUserRoles.ORG_ADMIN.value
2171 ]
2172 if not allowed_org_ids:
2173 raise HTTPException(
2174 status_code=403,
2175 detail={"error": "Only proxy admins and organization admins can list users."},
2176 )
2178 # If client also sent organization_ids, intersect with allowed orgs
2179 if organization_ids:
2180 requested: Final = set(oid.strip() for oid in organization_ids.split(",") if oid.strip())
2181 intersection: Final = list(requested & set(allowed_org_ids))
2182 if not intersection:
2183 raise HTTPException(
2184 status_code=403,
2185 detail={"error": "You do not have org_admin access to the requested organization(s)."},
2186 )
2187 allowed_org_ids = intersection
2189 return ",".join(allowed_org_ids)
2192_NO_SEARCH_WHERE: Final[Mapping[str, object]] = MappingProxyType({})
2195def _user_search_where(search: str | None) -> Mapping[str, object]:
2196 """Prisma predicate for `/user/list?search=`: user_id or user_email contains it, case-insensitive."""
2197 if not search:
2198 return _NO_SEARCH_WHERE
2199 search_where: Final[UserSearchWhere] = {
2200 "OR": (
2201 {"user_id": {"contains": search, "mode": "insensitive"}},
2202 {"user_email": {"contains": search, "mode": "insensitive"}},
2203 )
2204 }
2205 return search_where
2208@router.get(
2209 "/user/list",
2210 tags=["Internal User management"],
2211 dependencies=[Depends(user_api_key_auth)],
2212 response_model=UserListResponse,
2213)
2214async def get_users(
2215 role: str | None = fastapi.Query(default=None, description="Filter users by role"),
2216 user_ids: str | None = fastapi.Query(default=None, description="Get list of users by user_ids"),
2217 sso_user_ids: str | None = fastapi.Query(default=None, description="Get list of users by sso_user_id"),
2218 user_email: str | None = fastapi.Query(default=None, description="Filter users by partial email match"),
2219 search: str | None = fastapi.Query(
2220 default=None,
2221 description="Combined search: matches users whose 'user_id' or 'user_email' contains the value (case-insensitive).",
2222 ),
2223 team: str | None = fastapi.Query(default=None, description="Filter users by team id"),
2224 page: int = fastapi.Query(default=1, ge=1, description="Page number"),
2225 page_size: int = fastapi.Query(default=25, ge=1, le=100, description="Number of items per page"),
2226 sort_by: str | None = fastapi.Query(
2227 default=None,
2228 description="Column to sort by (e.g. 'user_id', 'user_email', 'created_at', 'spend')",
2229 ),
2230 sort_order: str = fastapi.Query(default="asc", description="Sort order ('asc' or 'desc')"),
2231 organization_ids: str | None = fastapi.Query(
2232 default=None,
2233 description="Filter users by organization membership. Comma-separated list of org IDs.",
2234 ),
2235 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
2236):
2237 """
2238 Get a paginated list of users with filtering and sorting options.
2240 Parameters:
2241 role: Optional[str]
2242 Filter users by role. Can be one of:
2243 - proxy_admin
2244 - proxy_admin_viewer
2245 - internal_user
2246 - internal_user_viewer
2247 user_ids: Optional[str]
2248 Get list of users by user_ids. Comma separated list of user_ids.
2249 sso_ids: Optional[str]
2250 Get list of users by sso_ids. Comma separated list of sso_ids.
2251 user_email: Optional[str]
2252 Filter users by partial email match
2253 search: Optional[str]
2254 Combined search: matches users whose user_id or user_email contains the value (case-insensitive)
2255 team: Optional[str]
2256 Filter users by team id. Will match if user has this team in their teams array.
2257 page: int
2258 The page number to return
2259 page_size: int
2260 The number of items per page
2261 sort_by: Optional[str]
2262 Column to sort by (e.g. 'user_id', 'user_email', 'created_at', 'spend')
2263 sort_order: Optional[str]
2264 Sort order ('asc' or 'desc')
2265 """
2266 from litellm.proxy.proxy_server import (
2267 prisma_client,
2268 proxy_logging_obj,
2269 user_api_key_cache,
2270 )
2272 if prisma_client is None: 2272 ↛ 2273line 2272 didn't jump to line 2273 because the condition on line 2272 was never true
2273 raise HTTPException(
2274 status_code=500,
2275 detail={"error": f"No db connected. prisma client={prisma_client}"},
2276 )
2278 # Server-side authorization: proxy admins see all, org admins see only their org(s)
2279 organization_ids = await _authorize_user_list_request(
2280 user_api_key_dict=user_api_key_dict,
2281 organization_ids=organization_ids,
2282 prisma_client=prisma_client,
2283 user_api_key_cache=user_api_key_cache,
2284 proxy_logging_obj=proxy_logging_obj,
2285 )
2287 # Calculate skip and take for pagination
2288 skip: Final = (page - 1) * page_size
2290 # Build where conditions based on provided parameters
2291 where_conditions: dict[str, object] = {}
2293 if role:
2294 where_conditions["user_role"] = role
2296 if user_ids and isinstance(user_ids, str):
2297 user_id_list: Final = [uid.strip() for uid in user_ids.split(",") if uid.strip()]
2298 if len(user_id_list) == 1: 2298 ↛ 2304line 2298 didn't jump to line 2304 because the condition on line 2298 was always true
2299 where_conditions["user_id"] = {
2300 "contains": user_id_list[0],
2301 "mode": "insensitive",
2302 }
2303 else:
2304 where_conditions["user_id"] = {
2305 "in": user_id_list,
2306 }
2308 if user_email is not None and isinstance(user_email, str):
2309 where_conditions["user_email"] = {
2310 "contains": user_email,
2311 "mode": "insensitive", # Case-insensitive search
2312 }
2314 if team is not None and isinstance(team, str):
2315 where_conditions["teams"] = {
2316 "has": team # Array contains for string arrays in Prisma
2317 }
2319 if sso_user_ids is not None and isinstance(sso_user_ids, str):
2320 sso_id_list: Final = [sid.strip() for sid in sso_user_ids.split(",") if sid.strip()]
2321 where_conditions["sso_user_id"] = {
2322 "in": sso_id_list,
2323 }
2325 if organization_ids:
2326 org_id_list: Final = [oid.strip() for oid in organization_ids.split(",") if oid.strip()]
2327 if org_id_list: 2327 ↛ 2331line 2327 didn't jump to line 2331 because the condition on line 2327 was always true
2328 where_conditions["organization_memberships"] = {"some": {"organization_id": {"in": org_id_list}}}
2330 ## Filter any none fastapi.Query params - e.g. where_conditions: {'user_email': {'contains': Query(None), 'mode': 'insensitive'}, 'teams': {'has': Query(None)}}
2331 where: Final[Mapping[str, object]] = {
2332 key: value
2333 for key, value in (*where_conditions.items(), *_user_search_where(search).items())
2334 if value is not None
2335 }
2337 # Build order_by conditions
2339 order_by: Final[dict[str, str] | None] = (
2340 _validate_sort_params(sort_by, sort_order) if sort_by is not None and isinstance(sort_by, str) else None
2341 )
2343 users: Final[Sequence[prisma_models.LiteLLM_UserTable]] = await UserRepository(prisma_client).table.find_many(
2344 where=where,
2345 skip=skip,
2346 take=page_size,
2347 order=(order_by if order_by else {"created_at": "desc"}), # Default to created_at desc if no sort specified
2348 )
2350 # Get total count of user rows
2351 total_count: Final[int] = await UserRepository(prisma_client).table.count(where=where)
2353 # Get key count for each user
2354 user_key_counts: Final = await get_user_key_counts(prisma_client, [user.user_id for user in users])
2356 verbose_proxy_logger.debug("Total count of users: %s", total_count)
2358 # Calculate total pages
2359 total_pages: Final = -(-total_count // page_size) # Ceiling division
2361 # Prepare response
2362 user_list: list[LiteLLM_UserTableWithKeyCount] = []
2363 for user in users:
2364 user_dump = user.model_dump()
2365 user_dump["metadata"] = _redact_scim_enterprise_metadata(user_dump.get("metadata"))
2366 user_list.append(
2367 LiteLLM_UserTableWithKeyCount.model_validate(
2368 {**user_dump, "key_count": user_key_counts.get(user.user_id, 0)}
2369 )
2370 )
2372 return {
2373 "users": user_list,
2374 "total": total_count,
2375 "page": page,
2376 "page_size": page_size,
2377 "total_pages": total_pages,
2378 }
2381@router.post(
2382 "/user/delete",
2383 tags=["Internal User management"],
2384 dependencies=[Depends(user_api_key_auth)],
2385)
2386@management_endpoint_wrapper
2387async def delete_user(
2388 data: DeleteUserRequest,
2389 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
2390 litellm_changed_by: str | None = Header(
2391 None,
2392 description="The litellm-changed-by header enables tracking of actions performed by authorized users on behalf of other users, providing an audit trail for accountability",
2393 ),
2394):
2395 """
2396 delete user and associated user keys
2398 ```
2399 curl --location 'http://0.0.0.0:4000/user/delete' \
2401 --header 'Authorization: Bearer sk-1234' \
2403 --header 'Content-Type: application/json' \
2405 --data-raw '{
2406 "user_ids": ["45e3e396-ee08-4a61-a88e-16b3ce7e0849"]
2407 }'
2408 ```
2410 Parameters:
2411 - user_ids: List[str] - The list of user id's to be deleted.
2412 """
2413 from litellm.proxy.management_endpoints.team_endpoints import (
2414 _cleanup_members_with_roles,
2415 )
2416 from litellm.proxy.management_helpers.audit_logs import (
2417 get_audit_log_changed_by,
2418 is_audit_logging_enabled,
2419 )
2420 from litellm.proxy.proxy_server import (
2421 create_audit_log_for_update,
2422 litellm_proxy_admin_name,
2423 prisma_client,
2424 proxy_logging_obj,
2425 user_api_key_cache,
2426 )
2428 if prisma_client is None: 2428 ↛ 2429line 2428 didn't jump to line 2429 because the condition on line 2428 was never true
2429 raise HTTPException(status_code=500, detail={"error": "No db connected"})
2431 if data.user_ids is None: 2431 ↛ 2432line 2431 didn't jump to line 2432 because the condition on line 2431 was never true
2432 raise HTTPException(status_code=400, detail={"error": "No user id passed in"})
2434 # Per-target authorization: the route-level gate accepts this call when
2435 # the caller is PROXY_ADMIN or an ORG_ADMIN of *any* org named in
2436 # request_data["organization_id"]/["organizations"]. That gate does NOT
2437 # cross-check data.user_ids against the caller's scope, so without this
2438 # loop an org-admin of org-A could delete users in org-B by supplying
2439 # {"user_ids": [victim_in_org_B], "organization_id": "org-A"}.
2440 caller_is_proxy_admin: Final = user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN.value
2441 caller_admin_org_ids: set[str] = set()
2442 if not caller_is_proxy_admin: 2442 ↛ 2443line 2442 didn't jump to line 2443 because the condition on line 2442 was never true
2443 caller_memberships: Final[Sequence[prisma_models.LiteLLM_OrganizationMembership]] = (
2444 await _organization_membership_table(prisma_client).find_many(
2445 where={
2446 "user_id": user_api_key_dict.user_id,
2447 "user_role": LitellmUserRoles.ORG_ADMIN.value,
2448 }
2449 )
2450 if user_api_key_dict.user_id
2451 else []
2452 )
2453 caller_admin_org_ids = {m.organization_id for m in caller_memberships if m.organization_id}
2454 if not caller_admin_org_ids:
2455 raise HTTPException(
2456 status_code=403,
2457 detail={"error": "Only PROXY_ADMIN or ORG_ADMIN users may delete users."},
2458 )
2460 # Batch-fetch target memberships once before the per-user loop. Avoids
2461 # an N+1 DB call when delete_user is called with a large user_ids list.
2462 target_org_ids_by_user: Final[dict[str, set[str]]] = {}
2463 if not caller_is_proxy_admin: 2463 ↛ 2464line 2463 didn't jump to line 2464 because the condition on line 2463 was never true
2464 all_target_memberships: Final = await _organization_membership_table(prisma_client).find_many(
2465 where={"user_id": {"in": data.user_ids}}
2466 )
2467 for m in all_target_memberships:
2468 if not m.organization_id:
2469 continue
2470 target_org_ids_by_user.setdefault(m.user_id, set()).add(m.organization_id)
2472 # check that all teams passed exist
2473 for user_id in data.user_ids:
2474 user_row = await UserRepository(prisma_client).table.find_unique(where={"user_id": user_id})
2476 if user_row is None: 2476 ↛ 2482line 2476 didn't jump to line 2482 because the condition on line 2476 was always true
2477 raise HTTPException(
2478 status_code=404,
2479 detail={"error": f"User not found, passed user_id={user_id}"},
2480 )
2482 if not caller_is_proxy_admin:
2483 target_org_ids = target_org_ids_by_user.get(user_id, set())
2484 # Org-admin may only delete users whose entire org membership is
2485 # within their admin scope. A target with ANY org outside the
2486 # caller's scope (or no org at all) requires PROXY_ADMIN.
2487 if not target_org_ids or not target_org_ids.issubset(caller_admin_org_ids):
2488 raise HTTPException(
2489 status_code=403,
2490 detail={
2491 "error": (
2492 f"User {user_id} is not within your admin scope. "
2493 "Only PROXY_ADMIN may delete users outside your "
2494 "administered organizations."
2495 )
2496 },
2497 )
2499 # we do this after the first for loop, since first for loop is for validation. we only want this inserted after validation passes
2500 if is_audit_logging_enabled():
2501 # make an audit log for each team deleted
2502 _user_row = user_row.model_dump_json(exclude_none=True)
2504 asyncio.create_task(
2505 create_audit_log_for_update(
2506 request_data=LiteLLM_AuditLogs(
2507 id=str(uuid.uuid4()),
2508 updated_at=datetime.now(timezone.utc),
2509 changed_by=get_audit_log_changed_by(
2510 litellm_changed_by=litellm_changed_by,
2511 user_api_key_dict=user_api_key_dict,
2512 litellm_proxy_admin_name=litellm_proxy_admin_name,
2513 ),
2514 changed_by_api_key=user_api_key_dict.api_key,
2515 table_name=LitellmTableNames.USER_TABLE_NAME,
2516 object_id=user_id,
2517 action="deleted",
2518 updated_values="{}",
2519 before_value=_user_row,
2520 )
2521 )
2522 )
2524 ## CLEANUP MEMBERS_WITH_ROLES
2525 fetch_all_teams: Sequence[prisma_models.LiteLLM_TeamTable] = await TeamRepository(
2526 prisma_client
2527 ).table.find_many(where={"team_id": {"in": user_row.teams}})
2528 teams_to_update: list[tuple[str, str]] = []
2529 for team in fetch_all_teams:
2530 removed_team_members, new_team_members = _cleanup_members_with_roles(
2531 existing_team_row=LiteLLM_TeamTable.model_validate(team.model_dump()),
2532 data=TeamMemberDeleteRequest(
2533 team_id=team.team_id,
2534 user_id=user_row.user_id,
2535 user_email=user_row.user_email,
2536 ),
2537 )
2538 if removed_team_members:
2539 _db_new_team_members: list[dict] = [m.model_dump() for m in new_team_members]
2540 teams_to_update.append((team.team_id, json.dumps(_db_new_team_members)))
2542 ## update teams
2544 for team_id, members_with_roles in teams_to_update:
2545 await TeamRepository(prisma_client).table.update(
2546 where={"team_id": team_id},
2547 data={"members_with_roles": members_with_roles},
2548 )
2549 # End of Audit logging
2551 ## DELETE ASSOCIATED KEYS
2552 key_filter: Final[_UserIdInFilter] = {"user_id": {"in": data.user_ids}}
2553 keys_to_delete: Final = await _verification_token_table(prisma_client).find_many(where=key_filter)
2554 hashed_tokens_to_delete: Final = tuple(key.token for key in keys_to_delete)
2555 jwt_mapping_cache_keys: Final = await get_jwt_key_mapping_cache_keys_for_tokens(
2556 hashed_tokens=hashed_tokens_to_delete,
2557 prisma_client=prisma_client,
2558 )
2559 await _verification_token_table(prisma_client).delete_many(where=key_filter)
2560 if keys_to_delete: 2560 ↛ 2561line 2560 didn't jump to line 2561 because the condition on line 2560 was never true
2561 KeyManagementEventHooks.create_key_deleted_audit_logs(
2562 keys_being_deleted=keys_to_delete,
2563 user_api_key_dict=user_api_key_dict,
2564 litellm_changed_by=litellm_changed_by,
2565 )
2566 await delete_cache_key_objects(
2567 hashed_tokens=hashed_tokens_to_delete,
2568 user_api_key_cache=user_api_key_cache,
2569 proxy_logging_obj=proxy_logging_obj,
2570 )
2571 await evict_and_broadcast(cache_keys=jwt_mapping_cache_keys, user_api_key_cache=user_api_key_cache)
2573 ## DELETE ASSOCIATED INVITATION LINKS
2574 await _invitation_link_table(prisma_client).delete_many(
2575 where={
2576 "OR": [
2577 {"user_id": {"in": data.user_ids}},
2578 {"created_by": {"in": data.user_ids}},
2579 {"updated_by": {"in": data.user_ids}},
2580 ]
2581 }
2582 )
2584 ## DELETE ASSOCIATED ORGANIZATION MEMBERSHIPS
2585 await _organization_membership_table(prisma_client).delete_many(where={"user_id": {"in": data.user_ids}})
2587 ## DELETE ASSOCIATED TEAM MEMBERSHIPS
2588 await _team_membership_table(prisma_client).delete_many(where={"user_id": {"in": data.user_ids}})
2590 ## DELETE USERS
2591 deleted_users: Final = await _user_table(prisma_client).delete_many(where={"user_id": {"in": data.user_ids}})
2592 await evict_and_broadcast(cache_keys=tuple(data.user_ids), user_api_key_cache=user_api_key_cache)
2594 return deleted_users
2597async def add_internal_user_to_organization(
2598 user_id: str,
2599 organization_id: str,
2600 user_role: LitellmUserRoles,
2601) -> "prisma_models.LiteLLM_OrganizationMembership":
2602 """
2603 Helper function to add an internal user to an organization
2605 Adds the user to LiteLLM_OrganizationMembership table
2607 - Checks if organization_id exists
2609 Raises:
2610 - Exception if database not connected
2611 - Exception if user_id or organization_id not found
2612 """
2613 from litellm.proxy.proxy_server import prisma_client
2615 if prisma_client is None:
2616 raise Exception("Database not connected")
2618 try:
2619 # Check if organization_id exists
2620 organization_row: Final = await _organization_table(prisma_client).find_unique(
2621 where={"organization_id": organization_id}
2622 )
2623 if organization_row is None:
2624 raise Exception(f"Organization not found, passed organization_id={organization_id}")
2626 # Create a new organization membership entry
2627 new_membership: Final[prisma_models.LiteLLM_OrganizationMembership] = await OrganizationMembershipRepository(
2628 prisma_client
2629 ).table.create(
2630 data={
2631 "user_id": user_id,
2632 "organization_id": organization_id,
2633 "user_role": user_role,
2634 # Note: You can also set budget within an organization if needed
2635 }
2636 )
2638 return new_membership
2639 except Exception as e:
2640 raise Exception(f"Failed to add user to organization: {e}")
2643async def _resolve_org_filter_for_user_search(
2644 user_api_key_dict: UserAPIKeyAuth,
2645 team_id: str | None,
2646 prisma_client: "PrismaClient | None",
2647 user_api_key_cache: "UserApiKeyCache",
2648 proxy_logging_obj: "ProxyLogging | None",
2649) -> list[str] | None:
2650 """
2651 Return a list of org IDs to filter by, or ``None`` for no filter.
2653 Reads the ``scope_user_search_to_org`` UI-setting flag and applies
2654 role-based access rules when the flag is ON.
2655 """
2656 from litellm.proxy.ui_crud_endpoints.proxy_setting_endpoints import (
2657 get_ui_settings_cached,
2658 )
2660 ui_settings: Final = await get_ui_settings_cached()
2661 if not ui_settings.get("scope_user_search_to_org", False):
2662 return None # flag OFF — no filtering
2664 if _user_has_admin_view(user_api_key_dict):
2665 return None # proxy admin — see everything
2667 # Try to resolve org admin memberships
2668 caller_user = None
2669 if user_api_key_dict.user_id is not None:
2670 try:
2671 caller_user = await get_user_object(
2672 user_id=user_api_key_dict.user_id,
2673 prisma_client=prisma_client,
2674 user_api_key_cache=user_api_key_cache,
2675 user_id_upsert=False,
2676 proxy_logging_obj=proxy_logging_obj,
2677 )
2678 except ValueError:
2679 caller_user = None
2681 # Collect org IDs from ALL org memberships (any role, not just ORG_ADMIN).
2682 # This allows team admins who are org members to search users in their org.
2683 member_org_ids: list[str] = []
2684 if caller_user is not None:
2685 member_org_ids = [m.organization_id for m in (caller_user.organization_memberships or [])]
2687 if member_org_ids:
2688 return member_org_ids
2690 # Fall back to resolving via team_id (query param or from the caller's API key)
2691 resolved_team_id: Final = team_id or user_api_key_dict.team_id
2692 if resolved_team_id is not None:
2693 return await _resolve_team_org_filter(
2694 user_api_key_dict,
2695 resolved_team_id,
2696 prisma_client,
2697 user_api_key_cache,
2698 proxy_logging_obj,
2699 )
2701 raise HTTPException(
2702 status_code=403,
2703 detail={
2704 "error": "scope_user_search_to_org is enabled. Only proxy admins, organization admins, or team admins can search users."
2705 },
2706 )
2709async def _resolve_team_org_filter(
2710 user_api_key_dict: UserAPIKeyAuth,
2711 team_id: str,
2712 prisma_client: "PrismaClient | None",
2713 user_api_key_cache: "UserApiKeyCache",
2714 proxy_logging_obj: "ProxyLogging | None",
2715) -> list[str]:
2716 """Look up the team and return its org as a filter list, or raise 403."""
2717 from litellm.proxy.management_endpoints.common_utils import _is_user_team_admin
2719 try:
2720 team_obj: Final = await get_team_object(
2721 team_id=team_id,
2722 prisma_client=prisma_client,
2723 user_api_key_cache=user_api_key_cache,
2724 proxy_logging_obj=proxy_logging_obj,
2725 )
2726 except HTTPException:
2727 raise HTTPException(
2728 status_code=403,
2729 detail={"error": f"scope_user_search_to_org is enabled but team '{team_id}' was not found."},
2730 )
2732 if not _is_user_team_admin(user_api_key_dict, team_obj):
2733 raise HTTPException(
2734 status_code=403,
2735 detail={"error": "scope_user_search_to_org is enabled. You must be an admin of this team to search users."},
2736 )
2738 if team_obj.organization_id:
2739 return [team_obj.organization_id]
2741 raise HTTPException(
2742 status_code=403,
2743 detail={
2744 "error": "scope_user_search_to_org is enabled and this team is not part of an organization. Contact your proxy admin to adjust this setting."
2745 },
2746 )
2749@router.get(
2750 "/user/filter/ui",
2751 tags=["Internal User management"],
2752 dependencies=[Depends(user_api_key_auth)],
2753 include_in_schema=False,
2754 responses={
2755 200: {"model": list[LiteLLM_UserTableFiltered]},
2756 },
2757)
2758async def ui_view_users(
2759 user_id: str | None = fastapi.Query(default=None, description="User ID in the request parameters"),
2760 user_email: str | None = fastapi.Query(default=None, description="User email in the request parameters"),
2761 search: str | None = fastapi.Query(
2762 default=None,
2763 description="Combined search: matches users whose 'user_id' or 'user_email' contains the value (case-insensitive).",
2764 ),
2765 team_id: str | None = fastapi.Query(
2766 default=None,
2767 description="Team ID — used when a team admin searches for users to add to their team",
2768 ),
2769 page: int = fastapi.Query(default=1, description="Page number for pagination", ge=1),
2770 page_size: int = fastapi.Query(default=50, description="Number of items per page", ge=1, le=100),
2771 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
2772):
2773 """
2774 Filter users based on partial match of user_id or email, or combined ``search``, with pagination.
2776 Behaviour depends on the ``scope_user_search_to_org`` UI-setting flag
2777 (stored in the ``litellm_uisettings`` table):
2779 * **Flag OFF (default):** any authenticated user can search all users.
2780 * **Flag ON:**
2781 - Proxy admins see all users.
2782 - Org admins see only users in their org(s).
2783 - Team admins for an org-bound team see users in that org.
2784 - Others receive a 403.
2785 """
2786 from litellm.proxy.proxy_server import (
2787 prisma_client,
2788 proxy_logging_obj,
2789 user_api_key_cache,
2790 )
2792 if prisma_client is None:
2793 raise HTTPException(status_code=500, detail={"error": "No db connected"})
2795 try:
2796 org_filter_ids: Final = await _resolve_org_filter_for_user_search(
2797 user_api_key_dict=user_api_key_dict,
2798 team_id=team_id,
2799 prisma_client=prisma_client,
2800 user_api_key_cache=user_api_key_cache,
2801 proxy_logging_obj=proxy_logging_obj,
2802 )
2804 # Calculate offset for pagination
2805 skip: Final = (page - 1) * page_size
2807 # Build where conditions based on provided parameters
2808 where_conditions: Final[prisma_types.LiteLLM_UserTableWhereInput] = {}
2810 if user_id:
2811 where_conditions["user_id"] = {
2812 "contains": user_id,
2813 "mode": "insensitive", # Case-insensitive search
2814 }
2816 if user_email:
2817 where_conditions["user_email"] = {
2818 "contains": user_email,
2819 "mode": "insensitive", # Case-insensitive search
2820 }
2822 # Apply org filter when scope_user_search_to_org is ON and caller is not proxy admin
2823 if org_filter_ids is not None:
2824 where_conditions["organization_memberships"] = {"some": {"organization_id": {"in": org_filter_ids}}}
2826 where: Final[Mapping[str, object]] = { # mutable-ok: prisma serializes `where`, keep it a plain dict
2827 key: value
2828 for key, value in (*where_conditions.items(), *_user_search_where(search).items())
2829 if value is not None
2830 }
2832 # Query users with pagination and filters
2833 users: Final = await _user_table(prisma_client).find_many(
2834 where=where,
2835 skip=skip,
2836 take=page_size,
2837 order={"created_at": "desc"},
2838 )
2840 if not users:
2841 return []
2843 return [LiteLLM_UserTableFiltered.model_validate(user.model_dump()) for user in users]
2845 except HTTPException:
2846 raise
2847 except Exception as e:
2848 if PrismaDBExceptionHandler.is_database_service_unavailable_error_in_chain(e):
2849 verbose_proxy_logger.warning("Database unavailable during user search: %s", type(e).__name__)
2850 raise PrismaDBExceptionHandler.service_unavailable_proxy_exception(e) from e
2851 verbose_proxy_logger.exception("Error searching users: %s", e)
2852 raise HTTPException(status_code=500, detail=f"Error searching users: {e}")
2855# Using shared metric helper implementations from common_daily_activity
2858async def _resolve_user_email_metadata(
2859 prisma_client: "PrismaClient", records: Sequence[DailySpendRecord]
2860) -> dict[str, dict]:
2861 """Map each user_id on the page to its email/alias so the Usage dashboard can
2862 label the 'Spend Per User' chart with the email instead of the raw UUID."""
2863 user_ids: Final = {
2864 user_id for record in records if isinstance(user_id := getattr(record, "user_id", None), str) and user_id
2865 }
2866 if not user_ids: 2866 ↛ 2868line 2866 didn't jump to line 2868 because the condition on line 2866 was always true
2867 return {}
2868 users: Final = await _user_table(prisma_client).find_many(where={"user_id": {"in": list(user_ids)}})
2869 return {user.user_id: {"user_email": user.user_email, "user_alias": user.user_alias} for user in users}
2872@router.get(
2873 "/user/daily/activity",
2874 tags=["Budget & Spend Tracking", "Internal User management"],
2875 dependencies=[Depends(user_api_key_auth)],
2876 response_model=SpendAnalyticsPaginatedResponse,
2877)
2878@management_endpoint_wrapper
2879async def get_user_daily_activity(
2880 start_date: str | None = fastapi.Query(
2881 default=None,
2882 description="Start date in YYYY-MM-DD format",
2883 ),
2884 end_date: str | None = fastapi.Query(
2885 default=None,
2886 description="End date in YYYY-MM-DD format",
2887 ),
2888 model: str | None = fastapi.Query(
2889 default=None,
2890 description="Filter by specific model",
2891 ),
2892 api_key: str | None = fastapi.Query(
2893 default=None,
2894 description="Filter by specific API key",
2895 ),
2896 user_id: str | None = fastapi.Query(
2897 default=None,
2898 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.",
2899 ),
2900 page: int = fastapi.Query(default=1, description="Page number for pagination", ge=1),
2901 page_size: int = fastapi.Query(default=50, description="Items per page", ge=1, le=1000),
2902 timezone: int | None = fastapi.Query(
2903 default=None,
2904 description="Timezone offset in minutes from UTC (e.g., 480 for PST). "
2905 "Matches JavaScript's Date.getTimezoneOffset() convention.",
2906 ),
2907 include_current_utc_day: bool = fastapi.Query(
2908 default=False,
2909 description="When the range ends on the caller's current local day, extend it to "
2910 "today's UTC bucket so spend written after the caller's local midnight (in UTC "
2911 "terms) is included. Requires the timezone parameter. Historical ranges are "
2912 "never extended.",
2913 ),
2914 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
2915) -> SpendAnalyticsPaginatedResponse:
2916 """
2917 [BETA] This is a beta endpoint. It will change.
2919 Meant to optimize querying spend data for analytics for a user.
2921 Reads daily spend records that only ever accumulate and are never affected by budget
2922 resets. Their total can legitimately exceed the `spend` field returned by
2923 `/v2/user/info`, which is a running budget counter that every budget reset sets back
2924 to zero (or to the overage above `max_budget` when `budget_rollover` is enabled).
2926 Returns:
2927 (by date)
2928 - spend
2929 - prompt_tokens
2930 - completion_tokens
2931 - cache_read_input_tokens
2932 - cache_creation_input_tokens
2933 - total_tokens
2934 - api_requests
2935 - breakdown by model, api_key, provider
2936 """
2937 from litellm.proxy.proxy_server import prisma_client
2939 if prisma_client is None: 2939 ↛ 2940line 2939 didn't jump to line 2940 because the condition on line 2939 was never true
2940 raise HTTPException(
2941 status_code=500,
2942 detail={"error": CommonProxyErrors.db_not_connected_error.value},
2943 )
2945 if start_date is None or end_date is None:
2946 raise HTTPException(
2947 status_code=status.HTTP_400_BAD_REQUEST,
2948 detail={"error": "Please provide start_date and end_date"},
2949 )
2951 try:
2952 is_admin: Final = _user_has_admin_view(user_api_key_dict)
2954 if is_admin: 2954 ↛ 2957line 2954 didn't jump to line 2957 because the condition on line 2954 was always true
2955 entity_id = user_id # None means global view, otherwise filter by user
2956 else:
2957 caller_user_id: Final = require_caller_user_id_for_non_admin(user_api_key_dict)
2958 if user_id is None:
2959 user_id = caller_user_id
2960 if user_id != caller_user_id:
2961 raise HTTPException(
2962 status_code=status.HTTP_403_FORBIDDEN,
2963 detail={"error": "Non-admin users can only view their own spend data."},
2964 )
2965 entity_id = user_id
2967 return await get_daily_activity(
2968 prisma_client=prisma_client,
2969 table_name="litellm_dailyuserspend",
2970 entity_id_field="user_id",
2971 entity_id=entity_id,
2972 entity_metadata_field=None,
2973 start_date=start_date,
2974 end_date=end_date,
2975 model=model,
2976 api_key=api_key,
2977 page=page,
2978 page_size=page_size,
2979 timezone_offset_minutes=timezone,
2980 include_current_utc_day=include_current_utc_day,
2981 resolve_entity_metadata=lambda records: _resolve_user_email_metadata(prisma_client, records),
2982 )
2984 except HTTPException:
2985 raise
2986 except Exception as e:
2987 verbose_proxy_logger.exception("/spend/daily/analytics: Exception occured - %s", e)
2988 raise HTTPException(
2989 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
2990 detail={"error": f"Failed to fetch analytics: {e}"},
2991 )
2994@router.get(
2995 "/user/daily/activity/aggregated",
2996 tags=["Budget & Spend Tracking", "Internal User management"],
2997 dependencies=[Depends(user_api_key_auth)],
2998 response_model=SpendAnalyticsPaginatedResponse,
2999)
3000@management_endpoint_wrapper
3001async def get_user_daily_activity_aggregated(
3002 start_date: str | None = fastapi.Query(
3003 default=None,
3004 description="Start date in YYYY-MM-DD format",
3005 ),
3006 end_date: str | None = fastapi.Query(
3007 default=None,
3008 description="End date in YYYY-MM-DD format",
3009 ),
3010 model: str | None = fastapi.Query(
3011 default=None,
3012 description="Filter by specific model",
3013 ),
3014 api_key: str | None = fastapi.Query(
3015 default=None,
3016 description="Filter by specific API key",
3017 ),
3018 user_id: str | None = fastapi.Query(
3019 default=None,
3020 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.",
3021 ),
3022 timezone: int | None = fastapi.Query(
3023 default=None,
3024 description="Timezone offset in minutes from UTC (e.g., 480 for PST). "
3025 "Matches JavaScript's Date.getTimezoneOffset() convention.",
3026 ),
3027 include_current_utc_day: bool = fastapi.Query(
3028 default=False,
3029 description="When the range ends on the caller's current local day, extend it to "
3030 "today's UTC bucket so spend written after the caller's local midnight (in UTC "
3031 "terms) is included. Requires the timezone parameter. Historical ranges are "
3032 "never extended.",
3033 ),
3034 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
3035) -> SpendAnalyticsPaginatedResponse:
3036 """
3037 Aggregated analytics for a user's daily activity without pagination.
3038 Returns the same response shape as the paginated endpoint with page metadata set to single-page.
3040 Reads daily spend records that only ever accumulate and are never affected by budget
3041 resets. Their total can legitimately exceed the `spend` field returned by
3042 `/v2/user/info`, which is a running budget counter that every budget reset sets back
3043 to zero (or to the overage above `max_budget` when `budget_rollover` is enabled).
3044 """
3045 from litellm.proxy.proxy_server import prisma_client
3047 if prisma_client is None: 3047 ↛ 3048line 3047 didn't jump to line 3048 because the condition on line 3047 was never true
3048 raise HTTPException(
3049 status_code=500,
3050 detail={"error": CommonProxyErrors.db_not_connected_error.value},
3051 )
3053 if start_date is None or end_date is None:
3054 raise HTTPException(
3055 status_code=status.HTTP_400_BAD_REQUEST,
3056 detail={"error": "Please provide start_date and end_date"},
3057 )
3059 try:
3060 is_admin: Final = _user_has_admin_view(user_api_key_dict)
3062 if is_admin: 3062 ↛ 3065line 3062 didn't jump to line 3065 because the condition on line 3062 was always true
3063 entity_id = user_id # None means global view, otherwise filter by user
3064 else:
3065 caller_user_id: Final = require_caller_user_id_for_non_admin(user_api_key_dict)
3066 if user_id is None:
3067 user_id = caller_user_id
3068 if user_id != caller_user_id:
3069 raise HTTPException(
3070 status_code=status.HTTP_403_FORBIDDEN,
3071 detail={"error": "Non-admin users can only view their own spend data."},
3072 )
3073 entity_id = user_id
3075 return await get_daily_activity_aggregated(
3076 prisma_client=prisma_client,
3077 table_name="litellm_dailyuserspend",
3078 entity_id_field="user_id",
3079 entity_id=entity_id,
3080 entity_metadata_field=None,
3081 start_date=start_date,
3082 end_date=end_date,
3083 model=model,
3084 api_key=api_key,
3085 timezone_offset_minutes=timezone,
3086 include_current_utc_day=include_current_utc_day,
3087 )
3089 except HTTPException:
3090 raise
3091 except Exception as e:
3092 verbose_proxy_logger.exception("/user/daily/activity/aggregated: Exception occured - %s", e)
3093 raise HTTPException(
3094 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
3095 detail={"error": f"Failed to fetch analytics: {e}"},
3096 )