Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/management_endpoints/common_utils.py: 38%
236 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1import math
2from collections.abc import Mapping
3from types import MappingProxyType
4from typing import TYPE_CHECKING, Any, Final, Optional, Union
6from fastapi import HTTPException, status
7from pydantic import BaseModel
10# Defined above the `litellm.proxy.*` imports so the name is bound even when
11# this module is imported first through the proxy import cycle (CodeQL:
12# module-level cyclic import). Depends only on `math` + `HTTPException`.
13def validate_finite_spend(spend: float | None) -> None:
14 """Reject NaN/±inf spend before it reaches the DB / spend counter.
16 A non-finite spend would otherwise slip past `spend >= max_budget`
17 enforcement, since any comparison with NaN (and `-inf >= max_budget`)
18 is False, letting the entity keep spending past its configured budget.
19 """
20 if spend is not None and not math.isfinite(spend):
21 raise HTTPException(
22 status_code=400,
23 detail={"error": f"spend must be a finite number. Received: {spend}"},
24 )
27def validate_budget_duration(budget_duration: str | None, status_code: int = 400) -> None:
28 """Reject budget durations that can't be parsed, are non-positive, or
29 overflow date math, so a bad value can't be persisted and later crash the
30 budget reset job.
32 A non-positive duration also resolves to a reset time of "now", which leaves
33 the row permanently due: the reset job re-reads it every tick and, once
34 enough of them exist, they fill each batch and starve every other tenant's
35 reset.
36 """
37 from litellm.proxy.common_utils.timezone_utils import budget_duration_error
39 error: Final = budget_duration_error(budget_duration)
40 if error is not None:
41 raise HTTPException(status_code=status_code, detail={"error": error})
44from litellm._logging import verbose_proxy_logger
45from litellm.caching import DualCache
46from litellm.proxy._types import (
47 CommonProxyErrors,
48 KeyRequestBase,
49 LiteLLM_ManagementEndpoint_MetadataFields,
50 LiteLLM_ManagementEndpoint_MetadataFields_Premium,
51 LiteLLM_OrganizationTable,
52 LiteLLM_ProjectTable,
53 LiteLLM_TeamTable,
54 LiteLLM_UserTable,
55 LitellmUserRoles,
56 NewProjectRequest,
57 UpdateProjectRequest,
58 UserAPIKeyAuth,
59)
60from litellm.proxy._types import ( # noqa: F401 re-exported
61 user_api_key_has_admin_view as _user_has_admin_view,
62)
63from litellm.proxy.common_utils.timezone_utils import get_budget_reset_time
64from litellm.proxy.utils import _premium_user_check
65from litellm.repositories.team_repository import TeamRepository
66from litellm.types.utils import BudgetConfig
68if TYPE_CHECKING: 68 ↛ 69line 68 didn't jump to line 69 because the condition on line 68 was never true
69 from litellm.proxy._types import NewProjectRequest, UpdateProjectRequest
70 from litellm.proxy.utils import PrismaClient, ProxyLogging
73def validate_team_model_max_budget(
74 model_max_budget: Mapping[str, BudgetConfig] | None,
75 premium_user: bool,
76) -> None:
77 """Reject a team `model_max_budget` the limiter could not enforce (no duration, bad cap, tpm/rpm limits)."""
78 if not model_max_budget:
79 return
80 if premium_user is not True: 80 ↛ 87line 80 didn't jump to line 87 because the condition on line 80 was always true
81 raise HTTPException(
82 status_code=403,
83 detail={
84 "error": f"Setting model_max_budget on a team is an enterprise feature. {CommonProxyErrors.not_premium_user.value}"
85 },
86 )
87 for model_name, budget_config in model_max_budget.items():
88 if not model_name.strip():
89 raise HTTPException(
90 status_code=400,
91 detail={"error": "model_max_budget keys must be non-empty model names"},
92 )
93 max_budget = budget_config.max_budget
94 if max_budget is None or not math.isfinite(max_budget) or max_budget < 0:
95 raise HTTPException(
96 status_code=400,
97 detail={
98 "error": (
99 f"model_max_budget[{model_name!r}].max_budget must be a non-negative finite number. "
100 f"Received: {max_budget}"
101 )
102 },
103 )
104 if budget_config.budget_duration is None:
105 raise HTTPException(
106 status_code=400,
107 detail={"error": f"model_max_budget[{model_name!r}] requires a budget_duration, e.g. '1d' or '30d'"},
108 )
109 validate_budget_duration(budget_config.budget_duration)
110 if budget_config.tpm_limit is not None or budget_config.rpm_limit is not None:
111 raise HTTPException(
112 status_code=400,
113 detail={
114 "error": (
115 f"model_max_budget[{model_name!r}] tpm_limit/rpm_limit are not enforced on a team; "
116 "set per-model rate limits on the key instead"
117 )
118 },
119 )
122def require_caller_user_id_for_non_admin(
123 user_api_key_dict: UserAPIKeyAuth,
124) -> str:
125 """Return the caller's user_id, or raise 403 if missing.
127 Non-admin analytics endpoints scope queries by the caller's own user_id.
128 Service-account keys are deliberately created with user_id=None
129 (key_management_endpoints.py forces ``data.user_id = None`` at key
130 creation). Without this guard, that None value flows through to the
131 daily-activity builder, which treats ``entity_id is None`` as "no filter"
132 and returns every tenant's data.
134 Callers must check is_admin first; this helper is only valid on the
135 non-admin scoping branch.
136 """
137 if user_api_key_dict.user_id is None:
138 raise HTTPException(
139 status_code=status.HTTP_403_FORBIDDEN,
140 detail={
141 "error": (
142 "Service-account keys cannot query user analytics. Use a user-bound key, or call as a proxy admin."
143 )
144 },
145 )
146 return user_api_key_dict.user_id
149def _check_passthrough_routes_caller_permission(
150 data: BaseModel,
151 user_api_key_dict: UserAPIKeyAuth,
152 *,
153 entity: str = "key",
154) -> None:
155 """
156 Only proxy admins may set `allowed_passthrough_routes` (top-level or under
157 `metadata`) — it short-circuits the role-based route gate, so keys and teams
158 must be gated identically.
159 """
160 # view-only admins excluded by design; blocked upstream from writes anyway
161 if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN.value: 161 ↛ 163line 161 didn't jump to line 163 because the condition on line 161 was always true
162 return
163 if getattr(data, "allowed_passthrough_routes", None):
164 raise HTTPException(
165 status_code=403,
166 detail={"error": f"Only proxy admins can set `allowed_passthrough_routes` on a {entity}."},
167 )
168 metadata: Final = getattr(data, "metadata", None)
169 if isinstance(metadata, dict) and metadata.get("allowed_passthrough_routes"):
170 raise HTTPException(
171 status_code=403,
172 detail={"error": f"Only proxy admins can set `metadata.allowed_passthrough_routes` on a {entity}."},
173 )
176def _check_disable_global_guardrails_caller_permission(
177 disable_global_guardrails: bool | None,
178 metadata: Mapping[str, object] | None,
179 user_api_key_dict: UserAPIKeyAuth,
180 *,
181 entity: str = "key",
182 existing_metadata: Mapping[str, object] | None = None,
183) -> None:
184 """
185 Only proxy admins may opt a key or team out of default-on guardrails, whether the
186 flag is top-level or under `metadata`. Re-sending a flag that is already stored is
187 not an opt-out, so non-admin edits of an already exempted object still go through.
188 """
189 if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN.value: 189 ↛ 191line 189 didn't jump to line 191 because the condition on line 189 was always true
190 return
191 requested: Final = bool(disable_global_guardrails) or (
192 metadata is not None and bool(metadata.get("disable_global_guardrails"))
193 )
194 if not requested:
195 return
196 if existing_metadata is not None and existing_metadata.get("disable_global_guardrails") is True:
197 return
198 raise HTTPException(
199 status_code=403,
200 detail={"error": f"Only proxy admins can set `disable_global_guardrails` on a {entity}."},
201 )
204def _is_user_team_admin(user_api_key_dict: UserAPIKeyAuth, team_obj: LiteLLM_TeamTable) -> bool:
205 for member in team_obj.members_with_roles:
206 if (member.user_id is not None and member.user_id == user_api_key_dict.user_id) and member.role == "admin":
207 return True
209 return False
212async def _is_user_org_admin_for_team(user_api_key_dict: UserAPIKeyAuth, team_obj: LiteLLM_TeamTable) -> bool:
213 """
214 Check if user is an org admin for the team's organization.
216 Returns True if:
217 - The team belongs to an organization, AND
218 - The user has org_admin role in that organization
219 """
220 if not team_obj.organization_id or not user_api_key_dict.user_id:
221 return False
223 from litellm.proxy.auth.auth_checks import get_user_object
224 from litellm.proxy.proxy_server import (
225 prisma_client,
226 proxy_logging_obj,
227 user_api_key_cache,
228 )
230 caller_user: Final = await get_user_object(
231 user_id=user_api_key_dict.user_id,
232 prisma_client=prisma_client,
233 user_api_key_cache=user_api_key_cache,
234 user_id_upsert=False,
235 proxy_logging_obj=proxy_logging_obj,
236 )
237 if caller_user is None:
238 return False
240 for m in caller_user.organization_memberships or []:
241 if m.organization_id == team_obj.organization_id and m.user_role == LitellmUserRoles.ORG_ADMIN.value:
242 return True
244 return False
247def _team_member_has_permission(
248 user_api_key_dict: UserAPIKeyAuth,
249 team_obj: LiteLLM_TeamTable,
250 permission: str,
251) -> bool:
252 """Check if a non-admin team member has a specific permission on a team."""
253 if not team_obj.team_member_permissions:
254 return False
255 if permission not in team_obj.team_member_permissions:
256 return False
257 for member in team_obj.members_with_roles:
258 if member.user_id is not None and member.user_id == user_api_key_dict.user_id:
259 return True
260 return False
263async def _user_has_admin_privileges(
264 user_api_key_dict: UserAPIKeyAuth,
265 prisma_client: Optional["PrismaClient"] = None,
266 user_api_key_cache: Optional["DualCache"] = None,
267 proxy_logging_obj: Optional["ProxyLogging"] = None,
268) -> bool:
269 """
270 Check if user has admin privileges (proxy admin, team admin, or org admin).
272 Args:
273 user_api_key_dict: User API key authentication object
274 prisma_client: Prisma client for database operations
275 user_api_key_cache: Cache for user API keys
276 proxy_logging_obj: Proxy logging object
278 Returns:
279 True if user is proxy admin, team admin for any team, or org admin for any organization
280 """
281 # Check if user is proxy admin
282 if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN:
283 return True
285 # If no database connection, can't check team/org admin status
286 if prisma_client is None or user_api_key_dict.user_id is None:
287 return False
289 # Get user object to check team and org admin status
290 from litellm.caching import DualCache as DualCacheImport
291 from litellm.proxy.auth.auth_checks import get_user_object
293 try:
294 user_obj: Final = await get_user_object(
295 user_id=user_api_key_dict.user_id,
296 prisma_client=prisma_client,
297 user_api_key_cache=user_api_key_cache or DualCacheImport(),
298 user_id_upsert=False,
299 proxy_logging_obj=proxy_logging_obj,
300 )
302 if user_obj is None:
303 return False
305 # Check if user is org admin for any organization
306 if user_obj.organization_memberships is not None:
307 for membership in user_obj.organization_memberships:
308 if membership.user_role == LitellmUserRoles.ORG_ADMIN.value:
309 return True
311 # Check if user is team admin for any team
312 if user_obj.teams is not None and len(user_obj.teams) > 0:
313 # Get all teams user is in
314 teams = await TeamRepository(prisma_client).table.find_many(where={"team_id": {"in": user_obj.teams}})
316 for team in teams:
317 team_obj = LiteLLM_TeamTable.model_validate(team.model_dump())
318 if _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj):
319 return True
321 except Exception as e:
322 # If there's an error checking, default to False for security
323 verbose_proxy_logger.debug("Error checking admin privileges for user %s: %s", user_api_key_dict.user_id, e)
324 return False
326 return False
329def _org_admin_can_invite_user(
330 admin_user_obj: LiteLLM_UserTable,
331 target_user_obj: LiteLLM_UserTable,
332) -> bool:
333 """
334 Check if an org admin can invite the target user.
335 Target user must be in at least one org where the admin has org admin role.
337 Args:
338 admin_user_obj: The admin user's full object (from get_user_object)
339 target_user_obj: The target user's full object (from get_user_object)
341 Returns:
342 True if target user is in an org where admin has org admin role
343 """
344 if admin_user_obj.organization_memberships is None:
345 return False
346 admin_org_ids: Final = {
347 m.organization_id
348 for m in admin_user_obj.organization_memberships
349 if m.user_role == LitellmUserRoles.ORG_ADMIN.value
350 }
351 if not admin_org_ids:
352 return False
353 if target_user_obj.organization_memberships is None:
354 return False
355 target_org_ids: Final = {m.organization_id for m in target_user_obj.organization_memberships}
356 return bool(admin_org_ids & target_org_ids)
359async def _team_admin_can_invite_user(
360 user_api_key_dict: UserAPIKeyAuth,
361 admin_user_obj: LiteLLM_UserTable,
362 target_user_obj: LiteLLM_UserTable,
363 prisma_client: "PrismaClient",
364) -> bool:
365 """
366 Check if a team admin can invite the target user.
367 Target user must be in at least one team where the admin has team admin role.
369 Args:
370 user_api_key_dict: The admin user's API key auth object
371 admin_user_obj: The admin user's full object (from get_user_object)
372 target_user_obj: The target user's full object (from get_user_object)
373 prisma_client: Prisma client for database operations
375 Returns:
376 True if target user is in a team where admin has team admin role
377 """
378 if not admin_user_obj.teams or len(admin_user_obj.teams) == 0:
379 return False
380 if not target_user_obj.teams or len(target_user_obj.teams) == 0:
381 return False
383 teams: Final = await TeamRepository(prisma_client).table.find_many(where={"team_id": {"in": admin_user_obj.teams}})
384 admin_team_ids: Final = [
385 team.team_id
386 for team in teams
387 if _is_user_team_admin(
388 user_api_key_dict=user_api_key_dict,
389 team_obj=LiteLLM_TeamTable.model_validate(team.model_dump()),
390 )
391 ]
392 if not admin_team_ids:
393 return False
394 target_team_ids: Final = set(target_user_obj.teams)
395 return bool(set(admin_team_ids) & target_team_ids)
398async def admin_can_invite_user(
399 target_user_id: str,
400 user_api_key_dict: UserAPIKeyAuth,
401 prisma_client: Optional["PrismaClient"] = None,
402 user_api_key_cache: Optional["DualCache"] = None,
403 proxy_logging_obj: Optional["ProxyLogging"] = None,
404) -> bool:
405 """
406 Check if the admin can create an invitation for the target user.
407 - Proxy admins: can invite any user
408 - Org admins: can only invite users in their org(s)
409 - Team admins: can only invite users in their team(s)
411 Uses get_user_object for caching of both admin and target user objects.
413 Args:
414 target_user_id: The user_id of the user to invite
415 user_api_key_dict: The admin user's API key auth object
416 prisma_client: Prisma client for database operations
417 user_api_key_cache: Cache for user API keys
418 proxy_logging_obj: Proxy logging object
420 Returns:
421 True if user can invite the target user
422 """
423 if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN:
424 return True
426 if prisma_client is None or user_api_key_dict.user_id is None:
427 return False
429 from litellm.caching import DualCache as DualCacheImport
430 from litellm.proxy.auth.auth_checks import get_user_object
432 try:
433 cache: Final = user_api_key_cache or DualCacheImport()
434 admin_user_obj: Final = await get_user_object(
435 user_id=user_api_key_dict.user_id,
436 prisma_client=prisma_client,
437 user_api_key_cache=cache,
438 user_id_upsert=False,
439 proxy_logging_obj=proxy_logging_obj,
440 )
441 if admin_user_obj is None:
442 return False
444 target_user_obj: Final = await get_user_object(
445 user_id=target_user_id,
446 prisma_client=prisma_client,
447 user_api_key_cache=cache,
448 user_id_upsert=False,
449 proxy_logging_obj=proxy_logging_obj,
450 )
451 if target_user_obj is None:
452 return False
454 if _org_admin_can_invite_user(admin_user_obj, target_user_obj):
455 return True
457 if await _team_admin_can_invite_user(
458 user_api_key_dict=user_api_key_dict,
459 admin_user_obj=admin_user_obj,
460 target_user_obj=target_user_obj,
461 prisma_client=prisma_client,
462 ):
463 return True
465 return False
466 except Exception as e:
467 verbose_proxy_logger.debug("Error checking invite permission for user %s: %s", user_api_key_dict.user_id, e)
468 return False
471def _set_object_metadata_field(
472 object_data: Union[
473 LiteLLM_TeamTable,
474 KeyRequestBase,
475 LiteLLM_OrganizationTable,
476 LiteLLM_ProjectTable,
477 "NewProjectRequest",
478 "UpdateProjectRequest",
479 ],
480 field_name: str,
481 value: Any,
482) -> None:
483 """
484 Helper function to set metadata fields that require premium user checks
486 Args:
487 object_data: The team/key/organization/project data object to modify
488 field_name: Name of the metadata field to set
489 value: Value to set for the field
490 """
491 if field_name in LiteLLM_ManagementEndpoint_MetadataFields_Premium and value:
492 _premium_user_check(field_name)
494 object_data.metadata = object_data.metadata or {}
495 object_data.metadata[field_name] = value
498_TEAM_MEMBER_BUDGET_LIMIT_FIELDS: Final = (
499 "max_budget",
500 "soft_budget",
501 "max_parallel_requests",
502 "tpm_limit",
503 "rpm_limit",
504 "model_max_budget",
505 "budget_duration",
506 "allowed_models",
507 "temp_budget_increase",
508 "temp_budget_expiry",
509)
511_TEMP_BUDGET_FIELDS: Final = frozenset({"temp_budget_increase", "temp_budget_expiry"})
514MEMBER_BUDGET_PATCH_FIELDS: Final = MappingProxyType(
515 {
516 "max_budget_in_team": "max_budget",
517 "tpm_limit": "tpm_limit",
518 "rpm_limit": "rpm_limit",
519 "budget_duration": "budget_duration",
520 "allowed_models": "allowed_models",
521 "temp_budget_increase": "temp_budget_increase",
522 "temp_budget_expiry": "temp_budget_expiry",
523 }
524)
527def _prisma_value(value: object) -> object:
528 return list(value) if isinstance(value, tuple) else value
531def member_budget_patch(source: BaseModel) -> Mapping[str, object]:
532 """Map the per-member limit fields a request actually set to their budget-table
533 columns (merge-patch: a sent value updates, an explicit null clears, an absent
534 field is left untouched)."""
535 provided: Final = source.model_dump(exclude_unset=True)
536 return {
537 column: _prisma_value(provided[request_field])
538 for request_field, column in MEMBER_BUDGET_PATCH_FIELDS.items()
539 if request_field in provided
540 }
543def _is_set_budget_value(value: object) -> bool:
544 if value is None:
545 return False
546 if isinstance(value, list) and len(value) == 0: 546 ↛ 547line 546 didn't jump to line 547 because the condition on line 546 was never true
547 return False
548 return True
551def _has_meaningful_budget_limit(budget_values: Mapping[str, object]) -> bool:
552 """A budget is meaningful if at least one limit is actually set; an empty
553 list (no model restriction) and None both count as unset."""
554 return any(_is_set_budget_value(budget_values.get(field)) for field in _TEAM_MEMBER_BUDGET_LIMIT_FIELDS)
557async def _upsert_budget_and_membership(
558 tx,
559 *,
560 team_id: str,
561 user_id: str,
562 existing_budget_id: str | None,
563 user_api_key_dict: UserAPIKeyAuth,
564 budget_patch: Mapping[str, object],
565 team_default_budget_id: str | None = None,
566 shared_budget_ids: frozenset[str] | None = None,
567):
568 """
569 Apply a merge-patch of per-member budget fields to a team membership.
571 ``budget_patch`` holds only the budget columns the caller explicitly sent
572 (RFC 7396 semantics): a value sets the column, ``None`` clears it, and a
573 column that is absent from the dict is left untouched. Once the patch is
574 applied, if the budget has no meaningful limit left the member's private
575 budget is disconnected so they fall back to the team default.
577 ``team_default_budget_id`` is the team's shared default member budget id
578 (from team metadata.team_member_budget_id). When the membership still
579 points at it, we clone-on-write so editing one member's budget does not
580 mutate the shared default that every other member points at.
582 ``shared_budget_ids`` extends that protection to any other row more than one
583 membership points at, which a caller patching several members at once has
584 already counted; a row listed there is cloned rather than written in place.
585 A patch that only touches the temporary budget pair never copies permanent
586 limits into a new row, so the member keeps inheriting the live team default.
587 """
588 if not budget_patch:
589 return
591 write_data: Final = dict(budget_patch)
592 if "budget_duration" in write_data:
593 duration: Final = write_data["budget_duration"]
594 write_data["budget_reset_at"] = (
595 get_budget_reset_time(budget_duration=duration) if duration is not None else None
596 )
598 is_shared_default: Final = existing_budget_id is not None and (
599 existing_budget_id == team_default_budget_id or existing_budget_id in (shared_budget_ids or frozenset())
600 )
601 temp_only: Final = frozenset(write_data) <= _TEMP_BUDGET_FIELDS
603 async def _disconnect():
604 await tx.litellm_teammembership.update(
605 where={"user_id_team_id": {"user_id": user_id, "team_id": team_id}},
606 data={"litellm_budget_table": {"disconnect": True}},
607 )
609 if existing_budget_id is not None and not is_shared_default:
610 existing_budget: Final = await tx.litellm_budgettable.find_unique(where={"budget_id": existing_budget_id})
611 merged: Final = existing_budget.model_dump() if existing_budget is not None else {}
612 merged.update(write_data)
613 if not _has_meaningful_budget_limit(merged): 613 ↛ 614line 613 didn't jump to line 614 because the condition on line 613 was never true
614 await _disconnect()
615 return
616 await tx.litellm_budgettable.update(
617 where={"budget_id": existing_budget_id},
618 data={"updated_by": user_api_key_dict.user_id or "", **write_data},
619 )
620 return
622 source_row: Final = (
623 await tx.litellm_budgettable.find_unique(where={"budget_id": existing_budget_id})
624 if is_shared_default and not temp_only
625 else None
626 )
627 source: Final[Mapping[str, object]] = source_row.model_dump() if source_row is not None else MappingProxyType({})
629 create_data: Final[dict[str, object]] = { # mutable-ok: Prisma create payloads are dict-shaped
630 "created_by": user_api_key_dict.user_id or "",
631 "updated_by": user_api_key_dict.user_id or "",
632 **MappingProxyType(
633 {f: source[f] for f in _TEAM_MEMBER_BUDGET_LIMIT_FIELDS if _is_set_budget_value(source.get(f))}
634 ),
635 **write_data,
636 }
638 # Restarting an inherited window on an unrelated edit hands the member a free period.
639 carried: Final = source.get("budget_reset_at") if "budget_duration" not in budget_patch else None
640 if carried is not None: 640 ↛ 641line 640 didn't jump to line 641 because the condition on line 640 was never true
641 create_data["budget_reset_at"] = carried
642 if create_data.get("budget_reset_at") is None: 642 ↛ 645line 642 didn't jump to line 645 because the condition on line 642 was always true
643 create_data.pop("budget_reset_at", None)
645 if not _has_meaningful_budget_limit(create_data): 645 ↛ 646line 645 didn't jump to line 646 because the condition on line 645 was never true
646 if existing_budget_id is not None and not temp_only:
647 await _disconnect()
648 return
650 new_budget: Final = await tx.litellm_budgettable.create(
651 data=create_data,
652 include={"team_membership": True},
653 )
654 await tx.litellm_teammembership.upsert(
655 where={
656 "user_id_team_id": {
657 "user_id": user_id,
658 "team_id": team_id,
659 }
660 },
661 data={
662 "create": {
663 "user_id": user_id,
664 "team_id": team_id,
665 "litellm_budget_table": {
666 "connect": {"budget_id": new_budget.budget_id},
667 },
668 },
669 "update": {
670 "litellm_budget_table": {
671 "connect": {"budget_id": new_budget.budget_id},
672 },
673 },
674 },
675 )
678def _update_metadata_field(updated_kv: dict, field_name: str) -> None:
679 """
680 Helper function to update metadata fields that require premium user checks in the update endpoint
682 Args:
683 updated_kv: The key-value dict being used for the update
684 field_name: Name of the metadata field being updated
685 """
686 if field_name in LiteLLM_ManagementEndpoint_MetadataFields_Premium:
687 # The UI sends falsy defaults (False, [], {}) even when the user has not
688 # enabled any enterprise feature (see #20304, #30285); require a license
689 # only for a truthy value. The falsy value is still persisted below so a
690 # previously-set field can be cleared.
691 if updated_kv.get(field_name):
692 _premium_user_check()
694 if field_name in updated_kv and updated_kv[field_name] is not None: 694 ↛ exitline 694 didn't return from function '_update_metadata_field' because the condition on line 694 was always true
695 # remove field from updated_kv
696 _value: Final = updated_kv.pop(field_name)
697 if "metadata" in updated_kv and updated_kv["metadata"] is not None:
698 updated_kv["metadata"][field_name] = _value
699 else:
700 updated_kv["metadata"] = {field_name: _value}
703def _has_non_empty_value(value: object) -> bool:
704 """Check if a value has real content (not None, not empty list, not blank string)."""
705 if value is None:
706 return False
707 if isinstance(value, list) and len(value) == 0:
708 return False
709 if isinstance(value, str) and value.strip() == "":
710 return False
711 return True
714def _update_metadata_fields(updated_kv: dict) -> None:
715 """
716 Helper function to update all metadata fields (both premium and standard).
718 Args:
719 updated_kv: The key-value dict being used for the update
720 """
721 for field in LiteLLM_ManagementEndpoint_MetadataFields_Premium:
722 if field in updated_kv and updated_kv[field] is not None:
723 _update_metadata_field(updated_kv=updated_kv, field_name=field)
725 for field in LiteLLM_ManagementEndpoint_MetadataFields:
726 if field in updated_kv and updated_kv[field] is not None:
727 _update_metadata_field(updated_kv=updated_kv, field_name=field)