"""
TEAM MANAGEMENT

All /team management endpoints

/team/new
/team/info
/team/update
/team/delete
"""

import asyncio
import json
import math
import traceback
from collections.abc import Mapping, Sequence
from collections.abc import Set as AbstractSet
from datetime import datetime, timezone
from types import MappingProxyType
from typing import TYPE_CHECKING, Annotated, Final, NamedTuple, NoReturn, Protocol, TypeVar, cast

import fastapi
from fastapi import APIRouter, Depends, Header, HTTPException, Request, status
from pydantic import BaseModel, JsonValue
from typing_extensions import ReadOnly, TypedDict

import litellm
from litellm._logging import verbose_proxy_logger
from litellm._uuid import uuid
from litellm.integrations.prometheus import PrometheusLogger
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
from litellm.proxy._types import (
    UI_TEAM_ID,
    BlockTeamRequest,
    BudgetNewRequest,
    CommonProxyErrors,
    DeleteTeamRequest,
    LiteLLM_AuditLogs,
    LiteLLM_DeletedTeamTable,
    LiteLLM_ManagementEndpoint_MetadataFields,
    LiteLLM_ManagementEndpoint_MetadataFields_Premium,
    LiteLLM_ModelTable,
    LiteLLM_OrganizationTable,
    LiteLLM_OrganizationTableWithMembers,
    LiteLLM_TeamMembership,
    LiteLLM_TeamTable,
    LiteLLM_TeamTableCachedObj,
    LiteLLM_UserTable,
    LitellmTableNames,
    LitellmUserRoles,
    Member,
    NewTeamRequest,
    OrgMember,
    PatchTeamRequest,
    ProxyErrorTypes,
    ProxyException,
    ResetSpendRequest,
    SpecialManagementEndpointEnums,
    SpecialModelNames,
    SpecialProxyStrings,
    TeamAccessGroupModelGrant,
    TeamAddMemberResponse,
    TeamInfoResponseObject,
    TeamInfoResponseObjectTeamTable,
    TeamListResponseObject,
    TeamMemberAddRequest,
    TeamMemberDeleteRequest,
    TeamMemberUpdateRequest,
    TeamMemberUpdateResponse,
    TeamModelAddRequest,
    TeamModelDeleteRequest,
    UpdateTeamRequest,
    UserAPIKeyAuth,
)
from litellm.proxy.auth.auth_checks import (
    OrganizationNotFoundError,
    _cache_team_object,
    allowed_route_check_inside_route,
    can_org_access_model,
    delete_cache_key_objects,
    delete_cache_team_object,
    get_org_object,
    get_team_membership,
    get_team_object,
    get_user_object,
    invalidate_team_member_spend_state,
)
from litellm.proxy.auth.auth_utils import (
    enforce_batch_enqueued_token_limit_is_admin_only,
    enforce_output_token_estimates_are_admin_only,
)
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.common_utils.callback_utils import encrypt_callback_vars
from litellm.proxy.common_utils.json_merge_patch import apply_json_merge_patch
from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
from litellm.proxy.management_endpoints.common_daily_activity import (
    get_daily_activity_aggregated,
)
from litellm.proxy.management_endpoints.common_utils import (
    _check_passthrough_routes_caller_permission,
    _is_user_org_admin_for_team,
    _is_user_team_admin,
    _set_object_metadata_field,
    _team_member_has_permission,
    _update_metadata_fields,
    _upsert_budget_and_membership,
    _user_has_admin_view,
    validate_budget_duration,
)
from litellm.proxy.management_endpoints.organization_endpoints import (
    add_member_to_organization,
)
from litellm.proxy.management_endpoints.tag_management_endpoints import (
    get_daily_activity,
)
from litellm.proxy.management_helpers.access_group_team_sync import (
    TEAM_ADVISORY_LOCK_SQL,
    AccessGroupSyncTx,
    invalidate_access_group_caches,
    reconcile_team_access_group_membership,
    sync_team_access_group_membership,
)
from litellm.proxy.management_helpers.object_permission_utils import (
    _set_object_permission,
    enforce_all_proxy_mcp_servers_grant_is_admin_only,
    handle_update_object_permission_common,
)
from litellm.proxy.management_helpers.team_member_permission_checks import (
    TeamMemberPermissionChecks,
)
from litellm.proxy.management_helpers.team_metadata_validation import (
    TEAM_METADATA_SCHEMA_REGISTRY,
    validate_team_metadata_if_configured,
)
from litellm.proxy.management_helpers.utils import (
    MemberWriteTx,
    add_new_member,
    management_endpoint_wrapper,
)
from litellm.proxy.utils import PrismaClient, ProxyLogging, handle_exception_on_proxy
from litellm.repositories.budget_repository import BudgetRepository
from litellm.repositories.organization_repository import OrganizationRepository
from litellm.repositories.prisma_protocols import TableActions
from litellm.repositories.table_repositories import (
    AccessGroupRepository,
    DeletedTeamRepository,
    ModelTableRepository,
    OrganizationMembershipRepository,
    TeamMembershipRepository,
)
from litellm.repositories.team_repository import TeamRepository
from litellm.repositories.user_repository import UserRepository
from litellm.repositories.verification_token_repository import (
    VerificationTokenRepository,
)
from litellm.router import Router
from litellm.types.proxy.management_endpoints.common_daily_activity import (
    SpendAnalyticsPaginatedResponse,
)
from litellm.types.proxy.management_endpoints.team_endpoints import (
    BulkTeamMemberAddRequest,
    BulkTeamMemberAddResponse,
    BulkUpdateTeamMemberPermissionsRequest,
    BulkUpdateTeamMemberPermissionsResponse,
    GetTeamMemberPermissionsResponse,
    TeamIdSearchMatch,
    TeamListItem,
    TeamListResponse,
    TeamMemberAddResult,
    TeamMemberInfoResponse,
    TeamMetadataSchemaResponse,
    UpdateTeamMemberPermissionsRequest,
)

if TYPE_CHECKING:
    from prisma import Prisma
    from prisma import models as prisma_models

router: Final = APIRouter()

_DbRecordT = TypeVar("_DbRecordT")


class _TeamIdKeyCount(TypedDict):
    team_id: int


class _TeamIdGroupRow(TypedDict):
    team_id: str
    _count: _TeamIdKeyCount


def _as_object(value: object) -> object:
    return value


def _as_list(rows: Sequence[_DbRecordT]) -> list[_DbRecordT]:  # mutable-ok: pydantic list[...] fields reject Sequence
    return cast(  # cast-ok: prisma-client-py find_many returns a list; TableActions only widens it to Sequence
        "list[_DbRecordT]", rows
    )


class _UserIdRow(Protocol):
    @property
    def user_id(self) -> str | None: ...


def _user_id_rows_db(repo: UserRepository) -> "TableActions[_UserIdRow]":
    return repo.table


class _ModelDumpRow(Protocol):
    def model_dump(self) -> Mapping[str, object]: ...


class _TeamIdRow(Protocol):
    @property
    def team_id(self) -> str: ...


class _CacheableTeamRow(_TeamIdRow, _ModelDumpRow, Protocol): ...


class _ObjectPermissionRow(Protocol):
    @property
    def object_permission_id(self) -> str | None: ...


class _TeamAliasBudgetRow(Protocol):
    @property
    def team_alias(self) -> str | None: ...

    @property
    def budget_duration(self) -> str | None: ...


class _TeamBudgetRow(_TeamAliasBudgetRow, Protocol):
    metadata: Mapping[str, JsonValue] | None


class _AuditableTeamRow(Protocol):
    def json(self, *, exclude_none: bool = False) -> str: ...


class _RawTeamRow(_TeamIdRow, _ModelDumpRow, _ObjectPermissionRow, _TeamBudgetRow, _AuditableTeamRow, Protocol):
    @property
    def members_with_roles(
        self,
    ) -> Sequence[dict[str, object]] | None: ...  # mutable-ok: prisma deserializes this JSON column into plain dicts

    @property
    def organization_id(self) -> str | None: ...

    @property
    def max_budget(self) -> float | None: ...

    @property
    def soft_budget(self) -> float | None: ...

    @property
    def model_id(self) -> int | None: ...


def _raw_team_db(repo: TeamRepository) -> "TableActions[_RawTeamRow]":
    return cast(  # cast-ok: prisma types Json columns as str; the client hands back the deserialized value
        "TableActions[_RawTeamRow]", repo.table
    )


class _BudgetIdRow(Protocol):
    @property
    def budget_id(self) -> str: ...


class _BudgetWriteCall(Protocol):
    async def __call__(self, budget_obj: BudgetNewRequest, user_api_key_dict: UserAPIKeyAuth) -> _BudgetIdRow: ...


def _as_budget_write(fn: "_BudgetWriteCall") -> "_BudgetWriteCall":
    return fn


class _TeamFindManyArgs(TypedDict, total=False):
    take: int
    skip: int
    order: Mapping[str, str]
    cursor: Mapping[str, object]


class _TeamUiViewFilters(TypedDict, total=False):
    team_id: Mapping[str, str]
    team_alias: Mapping[str, str]


class _TeamIdInFilter(TypedDict, total=False):
    team_id: Mapping[str, Sequence[str]]


class _DeletedTeamsResult(TypedDict):
    deleted_teams: ReadOnly[Sequence[str]]


class _ErrorDetail(TypedDict):
    error: ReadOnly[str]


class _TeamCreateTx(AccessGroupSyncTx, Protocol):
    @property
    def litellm_teamtable(self) -> "TableActions[prisma_models.LiteLLM_TeamTable]": ...


class _MemberDeleteTx(Protocol):
    """The tables `/team/member_delete` reads while it holds the team's advisory lock.

    Reading them off the transaction keeps the whole endpoint on the one pooled connection
    it already checked out: a request that has the lock but still needs another connection
    can be starved by the lock waiters, which is a deadlock rather than a wait when enough
    of them hold the rest of the pool."""

    @property
    def litellm_usertable(self) -> "TableActions[prisma_models.LiteLLM_UserTable]": ...

    @property
    def litellm_verificationtoken(self) -> "TableActions[prisma_models.LiteLLM_VerificationToken]": ...


class _TeamDeleteTx(AccessGroupSyncTx, Protocol):
    async def execute_raw(self, query: str, *args: object) -> int: ...

    @property
    def litellm_teamtable(self) -> "TableActions[prisma_models.LiteLLM_TeamTable]": ...

    @property
    def litellm_teammembership(self) -> "TableActions[prisma_models.LiteLLM_TeamMembership]": ...


_STRIP_DELETED_TEAM_FROM_USERS_SQL: Final = """
UPDATE "LiteLLM_UserTable" SET teams = array_remove(teams, $1) WHERE $1 = ANY(teams)
"""

_INCLUDE_MODEL_TABLE: Final = MappingProxyType({"litellm_model_table": True})


def _team_db(prisma_client: PrismaClient | None) -> "TableActions[prisma_models.LiteLLM_TeamTable]":
    return TeamRepository(prisma_client).table


def _team_tx_db(tx: "Prisma") -> "TableActions[prisma_models.LiteLLM_TeamTable]":
    return cast(  # cast-ok: generated actions type Json columns as str; TableActions widens inputs to Mapping
        "TableActions[prisma_models.LiteLLM_TeamTable]", tx.litellm_teamtable
    )


def _team_membership_db(prisma_client: PrismaClient | None) -> "TableActions[prisma_models.LiteLLM_TeamMembership]":
    return TeamMembershipRepository(prisma_client).table


def _user_db(prisma_client: PrismaClient | None) -> "TableActions[prisma_models.LiteLLM_UserTable]":
    return UserRepository(prisma_client).table


def _model_db(prisma_client: PrismaClient | None) -> "TableActions[prisma_models.LiteLLM_ModelTable]":
    return ModelTableRepository(prisma_client).table


def _org_db(prisma_client: PrismaClient | None) -> "TableActions[prisma_models.LiteLLM_OrganizationTable]":
    return OrganizationRepository(prisma_client).table


def _org_membership_db(
    prisma_client: PrismaClient | None,
) -> "TableActions[prisma_models.LiteLLM_OrganizationMembership]":
    return OrganizationMembershipRepository(prisma_client).table


def _budget_db(prisma_client: PrismaClient | None) -> "TableActions[prisma_models.LiteLLM_BudgetTable]":
    return BudgetRepository(prisma_client).table


def _deleted_team_db(prisma_client: PrismaClient | None) -> "TableActions[prisma_models.LiteLLM_DeletedTeamTable]":
    return DeletedTeamRepository(prisma_client).table


def _access_group_db(prisma_client: PrismaClient | None) -> "TableActions[prisma_models.LiteLLM_AccessGroupTable]":
    return AccessGroupRepository(prisma_client).table


def _tokens_db(prisma_client: PrismaClient | None) -> "TableActions[prisma_models.LiteLLM_VerificationToken]":
    return VerificationTokenRepository(prisma_client).table


def _sanitize_for_log(value: object) -> str:
    """Strip CR/LF from user-controlled values to prevent log injection."""
    try:
        text = str(value)
    except Exception:
        text = repr(value)
    return text.replace("\r", "").replace("\n", "")


async def _refresh_cached_team(
    team_row: _CacheableTeamRow,
    user_api_key_cache: UserApiKeyCache,
    proxy_logging_obj: ProxyLogging,
) -> None:
    """
    Refresh the in-memory cached team object after a DB write.

    Every endpoint that mutates `litellm_teamtable` must call this so the
    cached `LiteLLM_TeamTableCachedObj` used by `common_checks` stays in
    sync. Without this, subsequent auth checks read a stale team and can
    403 on permissions the DB has already granted (or, symmetrically,
    keep granting permissions the DB has already revoked).

    `team_row` is the Prisma row returned by `update`/`find_unique` on
    `litellm_teamtable`. It is converted to `LiteLLM_TeamTableCachedObj`
    via `model_dump()` to match the cache shape `_cache_team_object`
    expects.
    """
    await _cache_team_object(
        team_id=team_row.team_id,
        team_table=LiteLLM_TeamTableCachedObj.model_validate(team_row.model_dump()),
        user_api_key_cache=user_api_key_cache,
        proxy_logging_obj=proxy_logging_obj,
    )


async def _verify_team_access(
    team_obj: LiteLLM_TeamTable,
    user_api_key_dict: UserAPIKeyAuth,
) -> None:
    """
    Verify the caller is authorized to manage the given team.

    Access is granted if:
    - Caller is a proxy admin, OR
    - Caller is an org admin for the team's organization, OR
    - Caller is a team admin of this team

    Raises HTTPException(403) otherwise.
    """
    if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN:
        return

    if _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj):
        return

    if await _is_user_org_admin_for_team(user_api_key_dict=user_api_key_dict, team_obj=team_obj):
        return

    raise HTTPException(
        status_code=status.HTTP_403_FORBIDDEN,
        detail="You do not have access to this team",
    )


class TeamMemberBudgetHandler:
    """Helper class to handle team member budget, RPM, and TPM limit operations"""

    # Metadata keys that are owned and set by the server. Callers must not be
    # able to inject or overwrite these via request payloads.
    SYSTEM_MANAGED_METADATA_KEYS = ("team_member_budget_id",)

    @staticmethod
    def strip_system_managed_metadata_keys(metadata: dict | None) -> None:
        """Remove server-owned metadata keys from a caller-supplied dict."""
        if not isinstance(metadata, dict):
            return
        for key in TeamMemberBudgetHandler.SYSTEM_MANAGED_METADATA_KEYS:
            metadata.pop(key, None)

    @staticmethod
    def should_create_budget(
        team_member_budget: float | None = None,
        team_member_rpm_limit: int | None = None,
        team_member_tpm_limit: int | None = None,
        team_member_budget_duration: str | None = None,
    ) -> bool:
        """Check if any team member limits are provided"""
        return any(
            [
                team_member_budget is not None,
                team_member_rpm_limit is not None,
                team_member_tpm_limit is not None,
                team_member_budget_duration is not None,
            ]
        )

    @staticmethod
    async def create_team_member_budget_table(
        data: NewTeamRequest | _TeamAliasBudgetRow,
        new_team_data_json: dict,
        user_api_key_dict: UserAPIKeyAuth,
        team_member_budget: float | None = None,
        team_member_rpm_limit: int | None = None,
        team_member_tpm_limit: int | None = None,
        team_member_budget_duration: str | None = None,
        explicitly_set_fields: AbstractSet[str] = frozenset(),
    ) -> dict:
        """Create team member budget table with provided limits.

        The team's own reset period is only inherited when the caller left the
        member duration out, so an explicit null means "never resets".
        """
        from litellm.proxy._types import BudgetNewRequest
        from litellm.proxy.management_endpoints.budget_management_endpoints import (
            new_budget,
        )

        if data.team_alias is not None:
            budget_id = f"team-{data.team_alias.replace(' ', '-')}-budget-{uuid.uuid4().hex}"
        else:
            budget_id = f"team-budget-{uuid.uuid4().hex}"

        # Create budget request with all provided limits
        budget_request: Final = BudgetNewRequest(
            budget_id=budget_id,
            budget_duration=(
                team_member_budget_duration
                if "team_member_budget_duration" in explicitly_set_fields
                else data.budget_duration or team_member_budget_duration
            ),
        )

        if team_member_budget is not None:
            budget_request.max_budget = team_member_budget
        if team_member_rpm_limit is not None:
            budget_request.rpm_limit = team_member_rpm_limit
        if team_member_tpm_limit is not None:
            budget_request.tpm_limit = team_member_tpm_limit
        if team_member_budget_duration is not None:
            budget_request.budget_duration = team_member_budget_duration

        team_member_budget_table: Final = await _as_budget_write(new_budget)(
            budget_obj=budget_request,
            user_api_key_dict=user_api_key_dict,
        )

        # Add team_member_budget_id as metadata field to team table
        if new_team_data_json.get("metadata") is None:
            new_team_data_json["metadata"] = {}
        new_team_data_json["metadata"]["team_member_budget_id"] = team_member_budget_table.budget_id

        # Remove team member fields from new_team_data_json
        TeamMemberBudgetHandler._clean_team_member_fields(new_team_data_json)

        return new_team_data_json

    @staticmethod
    async def upsert_team_member_budget_table(
        team_table: _TeamBudgetRow,
        user_api_key_dict: UserAPIKeyAuth,
        updated_kv: dict,
        team_member_budget: float | None = None,
        team_member_rpm_limit: int | None = None,
        team_member_tpm_limit: int | None = None,
        team_member_budget_duration: str | None = None,
        explicitly_set_fields: AbstractSet[str] = frozenset(),
    ) -> dict:
        """Upsert team member budget table with provided limits.

        A field the caller explicitly sent as null is written as null, so a
        team can keep a member budget while dropping its reset period.
        """
        from litellm.proxy._types import BudgetNewRequest
        from litellm.proxy.management_endpoints.budget_management_endpoints import (
            update_budget,
        )

        if team_table.metadata is None:
            team_table.metadata = {}

        team_member_budget_id: Final = team_table.metadata.get("team_member_budget_id")
        if team_member_budget_id is not None and isinstance(team_member_budget_id, str):
            # Budget exists - create update request with only provided values
            budget_request: Final = BudgetNewRequest(budget_id=team_member_budget_id)

            if team_member_budget is not None or "team_member_budget" in explicitly_set_fields:
                budget_request.max_budget = team_member_budget
            if team_member_rpm_limit is not None or "team_member_rpm_limit" in explicitly_set_fields:
                budget_request.rpm_limit = team_member_rpm_limit
            if team_member_tpm_limit is not None or "team_member_tpm_limit" in explicitly_set_fields:
                budget_request.tpm_limit = team_member_tpm_limit
            if team_member_budget_duration is not None or "team_member_budget_duration" in explicitly_set_fields:
                budget_request.budget_duration = team_member_budget_duration
                if team_member_budget_duration is None:
                    budget_request.budget_reset_at = None

            budget_row: Final = await _as_budget_write(update_budget)(
                budget_obj=budget_request,
                user_api_key_dict=user_api_key_dict,
            )
            verbose_proxy_logger.info(
                "Updated team member budget table: %s, with team_member_budget=%s, team_member_rpm_limit=%s, team_member_tpm_limit=%s",
                budget_row.budget_id,
                team_member_budget,
                team_member_rpm_limit,
                team_member_tpm_limit,
            )
            if updated_kv.get("metadata") is None:
                updated_kv["metadata"] = {}
            updated_kv["metadata"]["team_member_budget_id"] = budget_row.budget_id

        else:  # budget does not exist
            updated_kv = await TeamMemberBudgetHandler.create_team_member_budget_table(
                data=team_table,
                new_team_data_json=updated_kv,
                user_api_key_dict=user_api_key_dict,
                team_member_budget=team_member_budget,
                team_member_rpm_limit=team_member_rpm_limit,
                team_member_tpm_limit=team_member_tpm_limit,
                team_member_budget_duration=team_member_budget_duration,
                explicitly_set_fields=explicitly_set_fields,
            )

        # Remove team member fields from updated_kv
        TeamMemberBudgetHandler._clean_team_member_fields(updated_kv)
        return updated_kv

    @staticmethod
    def _clean_team_member_fields(data_dict: dict) -> None:
        """Remove team member fields from data dictionary"""
        data_dict.pop("team_member_budget", None)
        data_dict.pop("team_member_budget_duration", None)
        data_dict.pop("team_member_rpm_limit", None)
        data_dict.pop("team_member_tpm_limit", None)

    @staticmethod
    async def clear_team_member_budget_fields(
        team_table: _TeamBudgetRow,
        user_api_key_dict: "UserAPIKeyAuth",
        updated_kv: dict,
        explicitly_set_fields: set,
    ) -> dict:
        """Clear explicitly-nulled fields on the team member budget row."""
        from litellm.proxy._types import BudgetNewRequest
        from litellm.proxy.management_endpoints.budget_management_endpoints import (
            update_budget,
        )

        if team_table.metadata is None:
            team_table.metadata = {}

        team_member_budget_id: Final = team_table.metadata.get("team_member_budget_id")
        if team_member_budget_id is not None and isinstance(team_member_budget_id, str):
            budget_request: Final = BudgetNewRequest(budget_id=team_member_budget_id)
            if "team_member_budget" in explicitly_set_fields:
                budget_request.max_budget = None
            if "team_member_budget_duration" in explicitly_set_fields:
                budget_request.budget_duration = None
                budget_request.budget_reset_at = None
            if "team_member_rpm_limit" in explicitly_set_fields:
                budget_request.rpm_limit = None
            if "team_member_tpm_limit" in explicitly_set_fields:
                budget_request.tpm_limit = None
            await update_budget(
                budget_obj=budget_request,
                user_api_key_dict=user_api_key_dict,
            )

        TeamMemberBudgetHandler._clean_team_member_fields(updated_kv)
        return updated_kv

    @staticmethod
    async def backfill_team_member_budget_entries(
        team_id: str,
        members_with_roles: Sequence[Member | dict[str, object]],
        team_member_budget_id: str,
        prisma_client: PrismaClient,
    ) -> None:
        """
        Ensure every team member has a TeamMembership row linked to the
        team_member_budget.

        Called after team_member_budget is set/updated on a team. Creates
        rows for members who don't have one, and populates budget_id on
        existing rows where it is NULL. Rows with a non-NULL budget_id
        are left untouched, which preserves per-member overrides but also
        means rows pointing to a prior team-default budget_id are not
        migrated to the new one.
        """
        if not members_with_roles:
            return

        # Batch-fetch existing memberships for this team (avoids N+1 queries)
        existing_memberships: Final = await _team_membership_db(prisma_client).find_many(where={"team_id": team_id})
        existing_user_ids: Final = {m.user_id for m in existing_memberships}

        # Identify members with no existing membership row.
        # members_with_roles may contain Member instances or raw dicts depending
        # on how the team was fetched/deserialized.
        missing: Final = []
        for m in members_with_roles:
            user_id = m.get("user_id") if isinstance(m, dict) else m.user_id
            if user_id is not None and user_id not in existing_user_ids:
                missing.append(
                    {
                        "team_id": team_id,
                        "user_id": user_id,
                        "budget_id": team_member_budget_id,
                    }
                )

        if missing:
            await _team_membership_db(prisma_client).create_many(
                data=missing,
                skip_duplicates=True,  # safety net against concurrent races
            )
            verbose_proxy_logger.info(
                "Backfilled %d team_memberships for team %s with budget %s",
                len(missing),
                _sanitize_for_log(team_id),
                _sanitize_for_log(team_member_budget_id),
            )

        # Heal existing membership rows that predate the team_member_budget
        # configuration: populate budget_id where it is currently NULL.
        # Rows with an explicit budget_id (per-member override) are left alone.
        updated: Final = await _team_membership_db(prisma_client).update_many(
            where={"team_id": team_id, "budget_id": None},
            data={"budget_id": team_member_budget_id},
        )
        if updated:
            verbose_proxy_logger.info(
                "Populated budget_id on %d existing team_memberships for team %s with budget %s",
                updated,
                _sanitize_for_log(team_id),
                _sanitize_for_log(team_member_budget_id),
            )


def _get_default_team_param(field: str) -> object:
    """
    Returns a default value for the given field from litellm.default_team_params config.
    Returns None if no default is configured.

    For list fields containing enums (e.g. team_member_permissions), converts enum values to strings.
    """
    default_params: Final = litellm.default_team_params
    if default_params is None:
        return None
    if isinstance(default_params, dict):
        value = default_params.get(field)
    else:
        value = getattr(default_params, field, None)
    if value is None:
        return None
    # Convert enum values in lists to strings
    if isinstance(value, list):
        return [v.value if hasattr(v, "value") else v for v in value]
    return value


def _is_available_team(team_id: str, user_api_key_dict: UserAPIKeyAuth) -> bool:
    if litellm.default_internal_user_params is None:
        return False
    if "available_teams" in litellm.default_internal_user_params:
        return team_id in litellm.default_internal_user_params["available_teams"]
    return False


async def get_all_team_memberships(
    prisma_client: PrismaClient, team_ids: list[str], user_id: str | None = None
) -> list[LiteLLM_TeamMembership]:
    """Get all team memberships for a given user"""
    ## GET ALL MEMBERSHIPS ##
    where_obj: Final[dict[str, dict[str, list[str]]]] = {"team_id": {"in": team_ids}}
    if user_id is not None:
        where_obj["user_id"] = {"in": [user_id]}
    # if user_id is None:
    #     where_obj = {"team_id": {"in": team_id}}
    # else:
    #     where_obj = {"user_id": str(user_id), "team_id": {"in": team_id}}

    team_memberships: Final = await _team_membership_db(prisma_client).find_many(
        where=where_obj,
        include={"litellm_budget_table": True},
    )

    returned_tm: Final[list[LiteLLM_TeamMembership]] = []
    for tm in team_memberships:
        returned_tm.append(LiteLLM_TeamMembership.model_validate(tm.model_dump()))

    return returned_tm


def _check_team_model_specific_limits(
    teams: list[LiteLLM_TeamTable],
    data: NewTeamRequest | UpdateTeamRequest,
    entity_rpm_limit: int | None,
    entity_tpm_limit: int | None,
    entity_model_rpm_limit_dict: dict[str, int],
    entity_model_tpm_limit_dict: dict[str, int],
    entity_type: str,  # "organization"
) -> None:
    """
    Generic function to check if a team is allocating model specific limits.
    Raises an error if we're overallocating.
    """
    model_rpm_limit: Final = getattr(data, "model_rpm_limit", None) or (
        data.metadata.get("model_rpm_limit", None) if data.metadata else None
    )
    model_tpm_limit: Final = getattr(data, "model_tpm_limit", None) or (
        data.metadata.get("model_tpm_limit", None) if data.metadata else None
    )
    if model_rpm_limit is None and model_tpm_limit is None:
        return

    # get total model specific tpm/rpm limit
    model_specific_rpm_limit: Final[dict[str, int]] = {}
    model_specific_tpm_limit: Final[dict[str, int]] = {}

    for team in teams:
        if team.metadata and team.metadata.get("model_rpm_limit", None) is not None:
            for model, rpm_limit in team.metadata.get("model_rpm_limit", {}).items():
                model_specific_rpm_limit[model] = model_specific_rpm_limit.get(model, 0) + rpm_limit
        if team.metadata and team.metadata.get("model_tpm_limit", None) is not None:
            for model, tpm_limit in team.metadata.get("model_tpm_limit", {}).items():
                model_specific_tpm_limit[model] = model_specific_tpm_limit.get(model, 0) + tpm_limit

    if model_rpm_limit is not None:
        for model, rpm_limit in model_rpm_limit.items():
            if entity_rpm_limit is not None and model_specific_rpm_limit.get(model, 0) + rpm_limit > entity_rpm_limit:
                raise HTTPException(
                    status_code=400,
                    detail=f"Allocated RPM limit={model_specific_rpm_limit.get(model, 0)} + Team RPM limit={rpm_limit} is greater than {entity_type} RPM limit={entity_rpm_limit}",
                )
            elif entity_model_rpm_limit_dict:
                entity_model_specific_rpm_limit = entity_model_rpm_limit_dict.get(model)
                if (
                    entity_model_specific_rpm_limit
                    and model_specific_rpm_limit.get(model, 0) + rpm_limit > entity_model_specific_rpm_limit
                ):
                    raise HTTPException(
                        status_code=400,
                        detail=f"Allocated RPM limit={model_specific_rpm_limit.get(model, 0)} + Team RPM limit={rpm_limit} is greater than {entity_type} RPM limit={entity_model_specific_rpm_limit}",
                    )

    if model_tpm_limit is not None:
        for model, tpm_limit in model_tpm_limit.items():
            if entity_tpm_limit is not None and model_specific_tpm_limit.get(model, 0) + tpm_limit > entity_tpm_limit:
                raise HTTPException(
                    status_code=400,
                    detail=f"Allocated TPM limit={model_specific_tpm_limit.get(model, 0)} + Team TPM limit={tpm_limit} is greater than {entity_type} TPM limit={entity_tpm_limit}",
                )
            elif entity_model_tpm_limit_dict:
                entity_model_specific_tpm_limit = entity_model_tpm_limit_dict.get(model)
                if (
                    entity_model_specific_tpm_limit
                    and model_specific_tpm_limit.get(model, 0) + tpm_limit > entity_model_specific_tpm_limit
                ):
                    raise HTTPException(
                        status_code=400,
                        detail=f"Allocated TPM limit={model_specific_tpm_limit.get(model, 0)} + Team TPM limit={tpm_limit} is greater than {entity_type} TPM limit={entity_model_specific_tpm_limit}",
                    )


def _check_team_rpm_tpm_limits(
    teams: list[LiteLLM_TeamTable],
    data: NewTeamRequest | UpdateTeamRequest,
    entity_rpm_limit: int | None,
    entity_tpm_limit: int | None,
    entity_type: str,  # "organization"
) -> None:
    """
    Generic function to check if a team is allocating rpm/tpm limits.
    Raises an error if we're overallocating.
    """
    if teams is not None and len(teams) > 0:
        allocated_tpm = sum(team.tpm_limit for team in teams if team.tpm_limit is not None)
        allocated_rpm = sum(team.rpm_limit for team in teams if team.rpm_limit is not None)
    else:
        allocated_tpm = 0
        allocated_rpm = 0

    if (
        data.tpm_limit is not None
        and entity_tpm_limit is not None
        and data.tpm_limit + allocated_tpm > entity_tpm_limit
    ):
        raise HTTPException(
            status_code=400,
            detail=f"Allocated TPM limit={allocated_tpm} + Team TPM limit={data.tpm_limit} is greater than {entity_type} TPM limit={entity_tpm_limit}",
        )
    if (
        data.rpm_limit is not None
        and entity_rpm_limit is not None
        and data.rpm_limit + allocated_rpm > entity_rpm_limit
    ):
        raise HTTPException(
            status_code=400,
            detail=f"Allocated RPM limit={allocated_rpm} + Team RPM limit={data.rpm_limit} is greater than {entity_type} RPM limit={entity_rpm_limit}",
        )


def check_org_team_model_specific_limits(
    teams: list[LiteLLM_TeamTable],
    org_table: LiteLLM_OrganizationTable,
    data: NewTeamRequest | UpdateTeamRequest,
) -> None:
    """
    Check if the organization team is allocating model specific limits. If so, raise an error if we're overallocating.
    """

    # Get org limits from budget table if available
    entity_rpm_limit = None
    entity_tpm_limit = None
    entity_model_rpm_limit_dict = {}
    entity_model_tpm_limit_dict = {}

    if org_table.litellm_budget_table is not None:
        entity_rpm_limit = org_table.litellm_budget_table.rpm_limit
        entity_tpm_limit = org_table.litellm_budget_table.tpm_limit

    if org_table.metadata:
        entity_model_rpm_limit_dict = org_table.metadata.get("model_rpm_limit", {})
        entity_model_tpm_limit_dict = org_table.metadata.get("model_tpm_limit", {})

    _check_team_model_specific_limits(
        teams=teams,
        data=data,
        entity_rpm_limit=entity_rpm_limit,
        entity_tpm_limit=entity_tpm_limit,
        entity_model_rpm_limit_dict=entity_model_rpm_limit_dict,
        entity_model_tpm_limit_dict=entity_model_tpm_limit_dict,
        entity_type="organization",
    )


def check_org_team_rpm_tpm_limits(
    teams: list[LiteLLM_TeamTable],
    org_table: LiteLLM_OrganizationTable,
    data: NewTeamRequest | UpdateTeamRequest,
) -> None:
    """
    Check if the organization team is allocating rpm/tpm limits. If so, raise an error if we're overallocating.
    """
    # Get org limits from budget table if available
    entity_rpm_limit = None
    entity_tpm_limit = None

    if org_table.litellm_budget_table is not None:
        entity_rpm_limit = org_table.litellm_budget_table.rpm_limit
        entity_tpm_limit = org_table.litellm_budget_table.tpm_limit

    _check_team_rpm_tpm_limits(
        teams=teams,
        data=data,
        entity_rpm_limit=entity_rpm_limit,
        entity_tpm_limit=entity_tpm_limit,
        entity_type="organization",
    )


async def _check_org_team_limits(
    org_table: LiteLLM_OrganizationTable,
    data: NewTeamRequest | UpdateTeamRequest,
    prisma_client: PrismaClient,
) -> None:
    """
    Check organization team limits including:
    - Team budget vs organization's max_budget
    - Team models vs organization's allowed models
    - Guaranteed throughput limits (tpm/rpm) if applicable
    """

    # Validate team budget against organization's max_budget
    if (
        data.max_budget is not None
        and org_table.litellm_budget_table is not None
        and org_table.litellm_budget_table.max_budget is not None
        and data.max_budget > org_table.litellm_budget_table.max_budget
    ):
        raise HTTPException(
            status_code=400,
            detail={
                "error": f"Team max_budget ({data.max_budget}) exceeds organization's max_budget ({org_table.litellm_budget_table.max_budget}). Organization: {org_table.organization_id}"
            },
        )

    # Validate team models against organization's allowed models
    if data.models is not None and len(org_table.models) > 0:
        # If organization has 'all-proxy-models', skip validation as it allows all models
        if SpecialModelNames.all_proxy_models.value in org_table.models:
            pass
        else:
            for m in data.models:
                if m not in org_table.models:
                    raise HTTPException(
                        status_code=400,
                        detail={
                            "error": f"Model '{m}' not in organization's allowed models. Organization allowed models={org_table.models}. Organization: {org_table.organization_id}"
                        },
                    )

    # Validate team TPM/RPM against organization's TPM/RPM limits (direct comparison)
    if (
        data.tpm_limit is not None
        and org_table.litellm_budget_table is not None
        and org_table.litellm_budget_table.tpm_limit is not None
        and data.tpm_limit > org_table.litellm_budget_table.tpm_limit
    ):
        raise HTTPException(
            status_code=400,
            detail={
                "error": f"Team tpm_limit ({data.tpm_limit}) exceeds organization's tpm_limit ({org_table.litellm_budget_table.tpm_limit}). Organization: {org_table.organization_id}"
            },
        )

    if (
        data.rpm_limit is not None
        and org_table.litellm_budget_table is not None
        and org_table.litellm_budget_table.rpm_limit is not None
        and data.rpm_limit > org_table.litellm_budget_table.rpm_limit
    ):
        raise HTTPException(
            status_code=400,
            detail={
                "error": f"Team rpm_limit ({data.rpm_limit}) exceeds organization's rpm_limit ({org_table.litellm_budget_table.rpm_limit}). Organization: {org_table.organization_id}"
            },
        )

    # Check guaranteed throughput limits (only if applicable)
    rpm_limit_type: Final = getattr(data, "rpm_limit_type", None) or (
        data.metadata.get("rpm_limit_type", None) if data.metadata else None
    )
    tpm_limit_type: Final = getattr(data, "tpm_limit_type", None) or (
        data.metadata.get("tpm_limit_type", None) if data.metadata else None
    )

    if tpm_limit_type != "guaranteed_throughput" and rpm_limit_type != "guaranteed_throughput":
        return
    # get all organization teams
    # calculate allocated tpm/rpm limit
    # check if specified tpm/rpm limit is greater than allocated tpm/rpm limit

    teams: Final = await _team_db(prisma_client).find_many(
        where={"organization_id": org_table.organization_id},
    )

    # Convert teams to LiteLLM_TeamTable objects
    team_objs: Final[list[LiteLLM_TeamTable]] = []
    for team in teams:
        team_objs.append(LiteLLM_TeamTable.model_validate(team.model_dump()))

    check_org_team_model_specific_limits(
        teams=team_objs,
        org_table=org_table,
        data=data,
    )
    check_org_team_rpm_tpm_limits(
        teams=team_objs,
        org_table=org_table,
        data=data,
    )


async def _check_user_team_limits(
    data: NewTeamRequest | UpdateTeamRequest,
    user_api_key_dict: UserAPIKeyAuth,
    prisma_client: PrismaClient,
    user_api_key_cache: UserApiKeyCache,
) -> None:
    """
    Enforce the caller's personal limits when CREATING a standalone team.

    This validates the requested team budget / models / tpm / rpm against the
    caller's own limits, so a non-admin user cannot mint a brand-new team that
    is richer than themselves.

    Only used by /team/new for standalone teams (organization_id is None).
    /team/update does NOT call this — an existing team's admin is already
    authorized via _verify_team_access() and is not gated by their personal
    wallet. Org-scoped teams use _check_org_team_limits() instead.
    """
    # Validate team budget against user's max_budget
    if data.max_budget is not None and user_api_key_dict.user_id is not None:
        user_obj: Final = await get_user_object(
            user_id=user_api_key_dict.user_id,
            prisma_client=prisma_client,
            user_api_key_cache=user_api_key_cache,
            user_id_upsert=False,
        )

        if user_obj is not None and user_obj.max_budget is not None and data.max_budget > user_obj.max_budget:
            raise HTTPException(
                status_code=400,
                detail={
                    "error": f"max budget higher than user max. User max budget={user_obj.max_budget}. User role={user_api_key_dict.user_role}"
                },
            )

    # Validate team models against user's allowed models
    if data.models is not None and len(user_api_key_dict.models) > 0:
        for m in data.models:
            if m not in user_api_key_dict.models:
                raise HTTPException(
                    status_code=400,
                    detail={
                        "error": f"Model not in allowed user models. User allowed models={user_api_key_dict.models}. User id={user_api_key_dict.user_id}"
                    },
                )

    # Validate team TPM/RPM against user's TPM/RPM limits
    if (
        data.tpm_limit is not None
        and user_api_key_dict.tpm_limit is not None
        and data.tpm_limit > user_api_key_dict.tpm_limit
    ):
        raise HTTPException(
            status_code=400,
            detail={
                "error": f"tpm limit higher than user max. User tpm limit={user_api_key_dict.tpm_limit}. User role={user_api_key_dict.user_role}"
            },
        )

    if (
        data.rpm_limit is not None
        and user_api_key_dict.rpm_limit is not None
        and data.rpm_limit > user_api_key_dict.rpm_limit
    ):
        raise HTTPException(
            status_code=400,
            detail={
                "error": f"rpm limit higher than user max. User rpm limit={user_api_key_dict.rpm_limit}. User role={user_api_key_dict.user_role}"
            },
        )


def _check_team_budget_update_authority(
    data: UpdateTeamRequest,
    user_api_key_dict: UserAPIKeyAuth,
    existing_team_max_budget: float | None,
) -> None:
    """
    Restrict who can grow a standalone team's spend ceiling on /team/update.

    A team admin (already authorized via _verify_team_access) may keep or lower
    the team budget, but only a proxy admin may grow it - by raising max_budget
    above the team's current value or by removing the cap (setting it to None).
    Setting a finite budget on a team that has no cap is a restriction and is
    allowed. Org-scoped teams are governed by _check_org_team_limits().
    """
    if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN:
        return
    if existing_team_max_budget is None:
        return

    budget_explicitly_set: Final = "max_budget" in (getattr(data, "model_fields_set", None) or set())
    if budget_explicitly_set and data.max_budget is None:
        raise HTTPException(
            status_code=403,
            detail={
                "error": f"Only a proxy admin can remove a team's max_budget. Team's current max_budget={existing_team_max_budget}."
            },
        )

    if data.max_budget is not None and data.max_budget > existing_team_max_budget:
        raise HTTPException(
            status_code=403,
            detail={
                "error": f"Only a proxy admin can raise a team's max_budget. Team's current max_budget={existing_team_max_budget}, requested={data.max_budget}."
            },
        )


def _should_auto_add_team_creator(
    user_api_key_dict: UserAPIKeyAuth,
    general_settings: Mapping[str, object],
) -> bool:
    if user_api_key_dict.user_id is None:
        return False
    if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN:
        return True
    return general_settings.get("disable_auto_add_proxy_admin_to_teams") is not True


#### TEAM MANAGEMENT ####
@router.post(
    "/team/new",
    tags=["team management"],
    dependencies=[Depends(user_api_key_auth)],
    response_model=LiteLLM_TeamTable,
)
@management_endpoint_wrapper
async def new_team(
    data: NewTeamRequest,
    http_request: Request,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
    litellm_changed_by: str | None = Header(
        None,
        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",
    ),
):
    """
    Allow users to create a new team. Apply user permissions to their team.

    👉 [Detailed Doc on setting team budgets](https://docs.litellm.ai/docs/proxy/team_budgets)


    Parameters:
    - team_alias: Optional[str] - User defined team alias
    - team_id: Optional[str] - The team id of the user. If none passed, we'll generate it.
    - members_with_roles: List[{"role": "admin" or "user", "user_id": "<user-id>"}] - A list of users and their roles in the team. Get user_id when making a new user via `/user/new`.
    - team_member_permissions: Optional[List[str]] - A list of routes that non-admin team members can access. example: ["/key/generate", "/key/update", "/key/delete"]
    - metadata: Optional[dict] - Metadata for team, store information for team. Example metadata = {"extra_info": "some info"}
    - model_rpm_limit: Optional[Dict[str, int]] - The RPM (Requests Per Minute) limit for this team - applied across all keys for this team.
    - model_tpm_limit: Optional[Dict[str, int]] - The TPM (Tokens Per Minute) limit for this team - applied across all keys for this team.
    - default_estimated_output_tokens: Optional[int] - Expected output tokens reserved for TPM limiting when a request omits max_tokens, for keys on this team that do not set their own. Positive integer.
    - default_estimated_output_tokens_per_model: Optional[Dict[str, int]] - Per-model override of the above. Example: {"gpt-4": 4096, "gpt-3.5-turbo": 1024}
    - mcp_rpm_limit: Optional[Dict[str, int]] - Per-MCP-server RPM limit for this team, keyed by MCP server name (alias if set, else the configured name). Example: {"github": 100, "slack": 200}. Applied across all keys for this team.
    - tpm_limit: Optional[int] - The TPM (Tokens Per Minute) limit for this team - all keys with this team_id will have at max this TPM limit
    - rpm_limit: Optional[int] - The RPM (Requests Per Minute) limit for this team - all keys associated with this team_id will have at max this RPM limit
    - rpm_limit_type: Optional[Literal["guaranteed_throughput", "best_effort_throughput"]] - The type of RPM limit enforcement. Use "guaranteed_throughput" to raise an error if overallocating RPM, or "best_effort_throughput" for best effort enforcement.
    - tpm_limit_type: Optional[Literal["guaranteed_throughput", "best_effort_throughput"]] - The type of TPM limit enforcement. Use "guaranteed_throughput" to raise an error if overallocating TPM, or "best_effort_throughput" for best effort enforcement.
    - max_budget: Optional[float] - The maximum budget allocated to the team - all keys for this team_id will have at max this max_budget
    - soft_budget: Optional[float] - The soft budget threshold for the team. If max_budget is set, soft_budget must be strictly lower than max_budget. Can be set independently if max_budget is not set.
    - budget_duration: Optional[str] - The duration of the budget for the team. Doc [here](https://docs.litellm.ai/docs/proxy/team_budgets)
    - models: Optional[list] - A list of models associated with the team - all keys for this team_id will have at most, these models. If empty, assumes all models are allowed.
    - blocked: bool - Flag indicating if the team is blocked or not - will stop all calls from keys with this team_id.
    - members: Optional[List] - Control team members via `/team/member/add` and `/team/member/delete`.
    - tags: Optional[List[str]] - Tags for [tracking spend](https://litellm.vercel.app/docs/proxy/enterprise#tracking-spend-for-custom-tags) and/or doing [tag-based routing](https://litellm.vercel.app/docs/proxy/tag_routing).
    - prompts: Optional[List[str]] - List of prompts that the team is allowed to use.
    - organization_id: Optional[str] - The organization id of the team. Default is None. Create via `/organization/new`.
    - model_aliases: Optional[dict] - Model aliases for the team. [Docs](https://docs.litellm.ai/docs/proxy/team_based_routing#create-team-with-model-alias)
    - guardrails: Optional[List[str]] - Guardrails for the team. [Docs](https://docs.litellm.ai/docs/proxy/guardrails)
    - policies: Optional[List[str]] - Policies for the team. [Docs](https://docs.litellm.ai/docs/proxy/guardrails/guardrail_policies)
    - disable_global_guardrails: Optional[bool] - Whether to disable global guardrails for the key.
    - object_permission: Optional[LiteLLM_ObjectPermissionBase] - team-specific object permission. Example - {"vector_stores": ["vector_store_1", "vector_store_2"], "agents": ["agent_1", "agent_2"], "agent_access_groups": ["dev_group"]}. IF null or {} then no object permission.
    - team_member_budget: Optional[float] - The maximum budget allocated to an individual team member.
    - team_member_budget_duration: Optional[str] - The duration of the budget for the team member. Doc [here](https://docs.litellm.ai/docs/proxy/team_budgets)
    - team_member_rpm_limit: Optional[int] - The RPM (Requests Per Minute) limit for individual team members.
    - team_member_tpm_limit: Optional[int] - The TPM (Tokens Per Minute) limit for individual team members.
    - team_member_key_duration: Optional[str] - The duration for a team member's key. e.g. "1d", "1w", "1mo"
    - allowed_passthrough_routes: Optional[List[str]] - List of allowed pass through routes for the team.
    - allowed_vector_store_indexes: Optional[List[dict]] - List of allowed vector store indexes for the key. Example - [{"index_name": "my-index", "index_permissions": ["write", "read"]}]. If specified, the key will only be able to use these specific vector store indexes. Create index, using `/v1/indexes` endpoint.
    - secret_manager_settings: Optional[dict] - Secret manager settings for the team. [Docs](https://docs.litellm.ai/docs/secret_managers/overview)
    - router_settings: Optional[UpdateRouterConfig] - team-specific router settings. Example - {"model_group_retry_policy": {"gpt-4": {"RateLimitErrorRetries": 5}}}. IF null or {} then no router settings.
    - access_group_ids: Optional[List[str]] - List of access group IDs to associate with the team. Access groups define which models the team can access. Example - ["access_group_1", "access_group_2"].
    - enforced_file_expires_after: Optional[dict] - Enforced file expiration policy for the team. Keys created under this team will inherit this policy for file uploads. Example - {"anchor": "created_at", "days": 30}.
    - enforced_batch_output_expires_after: Optional[dict] - Enforced batch output file expiration policy for the team. Keys created under this team will inherit this policy for batch output files. Example - {"anchor": "created_at", "days": 30}.
    - budget_limits: Optional[list] - List of concurrent budget windows for the team. 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"}].
    - default_team_member_models: Optional[List[str]] - Default models assigned to new team members when they join this team. Must be a subset of the team's models.

    Returns:
    - team_id: (str) Unique team id - used for tracking spend across multiple keys for same team id.

    _deprecated_params:
    - admins: list - A list of user_id's for the admin role
    - users: list - A list of user_id's for the user role

    Example Request:
    ```
    curl --location 'http://0.0.0.0:4000/team/new' \
    --header 'Authorization: Bearer sk-1234' \
    --header 'Content-Type: application/json' \
    --data '{
      "team_alias": "my-new-team_2",
      "members_with_roles": [{"role": "admin", "user_id": "user-1234"},
        {"role": "user", "user_id": "user-2434"}]
    }'

    ```

     ```
    curl --location 'http://0.0.0.0:4000/team/new' \
    --header 'Authorization: Bearer sk-1234' \
    --header 'Content-Type: application/json' \
    --data '{
                "team_alias": "QA Prod Bot",
                "max_budget": 0.000000001,
                "budget_duration": "1d"
            }'
    ```
    """
    try:
        from litellm.proxy.management_helpers.audit_logs import (
            get_audit_log_changed_by,
            is_audit_logging_enabled,
        )
        from litellm.proxy.proxy_server import (
            _license_check,
            create_audit_log_for_update,
            general_settings,
            litellm_proxy_admin_name,
            prisma_client,
            user_api_key_cache,
        )

        if prisma_client is None:
            raise HTTPException(status_code=500, detail={"error": "No db connected"})

        # Validate budget values are not negative
        if data.max_budget is not None and (not math.isfinite(data.max_budget) or data.max_budget < 0):
            raise HTTPException(
                status_code=400,
                detail={"error": f"max_budget must be a non-negative finite number. Received: {data.max_budget}"},
            )
        if data.team_member_budget is not None and (
            not math.isfinite(data.team_member_budget) or data.team_member_budget < 0
        ):
            raise HTTPException(
                status_code=400,
                detail={
                    "error": f"team_member_budget must be a non-negative finite number. Received: {data.team_member_budget}"
                },
            )
        if data.soft_budget is not None and (not math.isfinite(data.soft_budget) or data.soft_budget < 0):
            raise HTTPException(
                status_code=400,
                detail={"error": f"soft_budget must be a non-negative finite number. Received: {data.soft_budget}"},
            )

        validate_budget_duration(data.budget_duration)
        validate_budget_duration(data.team_member_budget_duration)

        if data.soft_budget is not None:
            if data.max_budget is not None:
                # If max_budget is set, soft_budget must be strictly lower than max_budget
                if data.soft_budget >= data.max_budget:
                    raise HTTPException(
                        status_code=400,
                        detail={
                            "error": f"soft_budget ({data.soft_budget}) must be strictly lower than max_budget ({data.max_budget})"
                        },
                    )

        enforce_output_token_estimates_are_admin_only(
            data=data,
            existing_metadata=None,
            user_api_key_dict=user_api_key_dict,
            entity="team",
        )
        enforce_batch_enqueued_token_limit_is_admin_only(
            data=data,
            existing_metadata=None,
            user_api_key_dict=user_api_key_dict,
            entity="team",
        )

        # Check if license is over limit
        total_teams: Final = await _team_db(prisma_client).count()
        if total_teams and _license_check.is_team_count_over_limit(team_count=total_teams):
            raise HTTPException(
                status_code=403,
                detail="License is over limit. Please contact support@berri.ai to upgrade your license.",
            )

        if data.team_id is None:
            data.team_id = str(uuid.uuid4())
        else:
            if data.team_id == UI_TEAM_ID:
                raise HTTPException(
                    status_code=400,
                    detail={
                        "error": f"team_id '{UI_TEAM_ID}' is reserved for LiteLLM UI dashboard sessions and cannot be used for a real team. Please use a different team id."
                    },
                )
            # Check if team_id exists already
            _existing_team_id: Final = await prisma_client.get_data(
                team_id=data.team_id, table_name="team", query_type="find_unique"
            )
            if _existing_team_id is not None:
                raise HTTPException(
                    status_code=400,
                    detail={"error": f"Team id = {data.team_id} already exists. Please use a different team id."},
                )

        if data.organization_id is None:
            default_organization_id: Final = _get_default_team_param("organization_id")
            if isinstance(default_organization_id, str):
                data.organization_id = default_organization_id

        # Apply defaults from litellm.default_team_params to null fields.
        # budget_duration alone distinguishes explicit null (a deliberate
        # never-resetting budget, which the default must not override) from omitted.
        for field in (
            "max_budget",
            "budget_duration",
            "tpm_limit",
            "rpm_limit",
            "team_member_permissions",
        ):
            if getattr(data, field, None) is None and (
                field != "budget_duration" or field not in data.model_fields_set
            ):
                default_value = _get_default_team_param(field)
                if default_value is not None:
                    setattr(data, field, default_value)

        # Legacy fallback: apply max_budget from default_team_settings (YAML config)
        # if still not set after checking default_team_params.
        if data.max_budget is None:
            if (
                isinstance(litellm.default_team_settings, list)
                and len(litellm.default_team_settings) > 0
                and isinstance(litellm.default_team_settings[0], dict)
            ):
                default_budget: Final = litellm.default_team_settings[0].get("max_budget")
                if default_budget is not None:
                    data.max_budget = default_budget

        # check org key limits - done here to handle inheriting org id from team
        if data.organization_id is not None and prisma_client is not None:
            try:
                org_table = await get_org_object(
                    org_id=data.organization_id,
                    user_api_key_cache=user_api_key_cache,
                    prisma_client=prisma_client,
                    include_budget_table=True,
                )
            except OrganizationNotFoundError:
                org_table = None
            if org_table is None:
                raise HTTPException(
                    status_code=400,
                    detail=f"Organization not found for organization_id={data.organization_id}",
                )

            await _check_org_team_limits(
                org_table=org_table,
                data=data,
                prisma_client=prisma_client,
            )

        if (
            user_api_key_dict.user_role is None or user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN
        ):  # don't restrict proxy admin
            # Only validate user budget/models/tpm/rpm for standalone teams (not org-scoped)
            # For org-scoped teams, validation is done by _check_org_team_limits()
            if data.organization_id is None:
                await _check_user_team_limits(
                    data=data,
                    user_api_key_dict=user_api_key_dict,
                    prisma_client=prisma_client,
                    user_api_key_cache=user_api_key_cache,
                )

        if _should_auto_add_team_creator(user_api_key_dict, general_settings):
            creating_user_in_list: Final = any(
                member.user_id == user_api_key_dict.user_id for member in data.members_with_roles
            )
            if not creating_user_in_list:
                data.members_with_roles.append(Member(role="admin", user_id=user_api_key_dict.user_id))

        _check_passthrough_routes_caller_permission(data, user_api_key_dict, entity="team")

        if isinstance(data.metadata, dict):
            TeamMemberBudgetHandler.strip_system_managed_metadata_keys(data.metadata)

        await validate_team_metadata_if_configured(
            operation="create",
            metadata=data.metadata,
            existing_metadata=None,
            team_id=data.team_id,
            team_alias=data.team_alias,
            user_api_key_dict=user_api_key_dict,
        )

        ## ADD TO MODEL TABLE
        _model_id = None
        if data.model_aliases is not None and isinstance(data.model_aliases, dict):
            litellm_modeltable: Final = LiteLLM_ModelTable(
                model_aliases=json.dumps(data.model_aliases),
                created_by=user_api_key_dict.user_id or litellm_proxy_admin_name,
                updated_by=user_api_key_dict.user_id or litellm_proxy_admin_name,
            )
            model_dict: Final = await _model_db(prisma_client).create({**litellm_modeltable.json(exclude_none=True)})

            _model_id = model_dict.id

        data_json = data.json()

        ## Handle Object Permission - MCP, Vector Stores etc.
        await enforce_all_proxy_mcp_servers_grant_is_admin_only(
            requested_mcp_servers=(data.object_permission.mcp_servers if data.object_permission is not None else None),
            existing_object_permission_id=None,
            is_proxy_admin=user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN,
            prisma_client=prisma_client,
        )
        data_json = await _set_object_permission(
            data_json=data_json,
            prisma_client=prisma_client,
        )

        if TeamMemberBudgetHandler.should_create_budget(
            team_member_budget=data.team_member_budget,
            team_member_rpm_limit=data.team_member_rpm_limit,
            team_member_tpm_limit=data.team_member_tpm_limit,
            team_member_budget_duration=data.team_member_budget_duration,
        ):
            data_json = await TeamMemberBudgetHandler.create_team_member_budget_table(
                data=data,
                new_team_data_json=data_json,
                user_api_key_dict=user_api_key_dict,
                team_member_budget=data.team_member_budget,
                team_member_rpm_limit=data.team_member_rpm_limit,
                team_member_tpm_limit=data.team_member_tpm_limit,
                team_member_budget_duration=data.team_member_budget_duration,
                explicitly_set_fields=data.model_fields_set,
            )

        ## ADD TO TEAM TABLE
        complete_team_data: Final = LiteLLM_TeamTable(
            **data_json,
            model_id=_model_id,
        )

        # Set Management Endpoint Metadata Fields
        for field in LiteLLM_ManagementEndpoint_MetadataFields_Premium:
            if getattr(data, field, None) is not None:
                _set_object_metadata_field(
                    object_data=complete_team_data,
                    field_name=field,
                    value=getattr(data, field),
                )

        for field in LiteLLM_ManagementEndpoint_MetadataFields:
            if getattr(data, field, None) is not None:
                _set_object_metadata_field(
                    object_data=complete_team_data,
                    field_name=field,
                    value=getattr(data, field),
                )

        # If budget_duration is set, set `budget_reset_at`
        if complete_team_data.budget_duration is not None:
            from litellm.proxy.common_utils.timezone_utils import get_budget_reset_time

            complete_team_data.budget_reset_at = get_budget_reset_time(
                budget_duration=complete_team_data.budget_duration,
            )

        # If budget_limits is set, initialize reset_at for each window
        if complete_team_data.budget_limits:
            from litellm.proxy.common_utils.timezone_utils import get_budget_reset_time

            initialized_windows: Final = []
            for window in complete_team_data.budget_limits:
                w = window if isinstance(window, dict) else window.model_dump()
                w["reset_at"] = get_budget_reset_time(budget_duration=w["budget_duration"]).isoformat()
                initialized_windows.append(w)
            complete_team_data.budget_limits = initialized_windows

        ## Add Team Member Budget Table
        members_with_roles: list[Member] = []
        if complete_team_data.members_with_roles is not None:
            members_with_roles = complete_team_data.members_with_roles
            complete_team_data.members_with_roles = []

        complete_team_data_dict = complete_team_data.model_dump(exclude_none=True)

        # Serialize router_settings to JSON (matching key creation pattern)
        router_settings_value: Final = getattr(data, "router_settings", None)
        router_settings_json: Final = (
            safe_dumps(router_settings_value) if router_settings_value is not None else safe_dumps({})
        )
        complete_team_data_dict["router_settings"] = router_settings_json

        if complete_team_data_dict.get("metadata") is not None:
            complete_team_data_dict["metadata"] = encrypt_callback_vars(complete_team_data_dict["metadata"])

        complete_team_data_dict = prisma_client.jsonify_team_object(db_data=complete_team_data_dict)
        team_creation_data: Final[Mapping[str, object]] = complete_team_data_dict

        tx: _TeamCreateTx
        async with prisma_client.db.tx() as tx:
            team_row: Final[prisma_models.LiteLLM_TeamTable] = await tx.litellm_teamtable.create(
                data=team_creation_data,
                include=_INCLUDE_MODEL_TABLE,
            )
            affected_access_groups: Final = await reconcile_team_access_group_membership(tx, team_row.team_id)

        await invalidate_access_group_caches(affected_access_groups)

        ## ADD TEAM ID TO USER TABLE ##
        team_member_add_request: Final = TeamMemberAddRequest(
            team_id=data.team_id,
            member=members_with_roles,
        )
        await _add_team_members_to_team(
            data=team_member_add_request,
            complete_team_data=team_row,
            prisma_client=prisma_client,
            user_api_key_dict=user_api_key_dict,
            litellm_proxy_admin_name=litellm_proxy_admin_name,
        )

        if is_audit_logging_enabled():
            _updated_values = complete_team_data.json(exclude_none=True)

            _updated_values = json.dumps(_updated_values, default=str)

            asyncio.create_task(
                create_audit_log_for_update(
                    request_data=LiteLLM_AuditLogs(
                        id=str(uuid.uuid4()),
                        updated_at=datetime.now(timezone.utc),
                        changed_by=get_audit_log_changed_by(
                            litellm_changed_by=litellm_changed_by,
                            user_api_key_dict=user_api_key_dict,
                            litellm_proxy_admin_name=litellm_proxy_admin_name,
                        ),
                        changed_by_api_key=user_api_key_dict.api_key,
                        table_name=LitellmTableNames.TEAM_TABLE_NAME,
                        object_id=data.team_id,
                        action="created",
                        updated_values=_updated_values,
                        before_value=None,
                    )
                )
            )

        try:
            return team_row.model_dump()
        except Exception:
            return team_row.dict()
    except Exception as e:
        raise handle_exception_on_proxy(e)


async def _create_team_update_audit_log(
    existing_team_row: _AuditableTeamRow,
    updated_kv: dict,
    team_id: str,
    litellm_changed_by: str | None,
    user_api_key_dict: UserAPIKeyAuth,
    litellm_proxy_admin_name: str,
) -> None:
    """
    Create an audit log entry for team update operations.

    Args:
        existing_team_row: The team row before the update
        updated_kv: Dictionary of updated key-value pairs
        team_id: The ID of the team being updated
        litellm_changed_by: Optional header indicating who made the change
        user_api_key_dict: User API key authentication details
        litellm_proxy_admin_name: Name of the proxy admin
    """
    from litellm.proxy.management_helpers.audit_logs import (
        create_audit_log_for_update,
        get_audit_log_changed_by,
    )

    _before_value = existing_team_row.json(exclude_none=True)
    _before_value = json.dumps(_before_value, default=str)
    _after_value: Final[str] = json.dumps(updated_kv, default=str)

    asyncio.create_task(
        create_audit_log_for_update(
            request_data=LiteLLM_AuditLogs(
                id=str(uuid.uuid4()),
                updated_at=datetime.now(timezone.utc),
                changed_by=get_audit_log_changed_by(
                    litellm_changed_by=litellm_changed_by,
                    user_api_key_dict=user_api_key_dict,
                    litellm_proxy_admin_name=litellm_proxy_admin_name,
                ),
                changed_by_api_key=user_api_key_dict.api_key,
                table_name=LitellmTableNames.TEAM_TABLE_NAME,
                object_id=team_id,
                action="updated",
                updated_values=_after_value,
                before_value=_before_value,
            )
        )
    )


async def _update_model_table(
    data: UpdateTeamRequest,
    model_id: int | None,
    prisma_client: PrismaClient,
    user_api_key_dict: UserAPIKeyAuth,
    litellm_proxy_admin_name: str,
) -> int | None:
    """
    Upsert model table and return the model id
    """
    ## UPSERT MODEL TABLE
    _model_id = model_id
    if data.model_aliases is not None and isinstance(data.model_aliases, dict):
        litellm_modeltable: Final = LiteLLM_ModelTable(
            model_aliases=json.dumps(data.model_aliases),
            created_by=user_api_key_dict.user_id or litellm_proxy_admin_name,
            updated_by=user_api_key_dict.user_id or litellm_proxy_admin_name,
        )
        if model_id is None:
            model_dict = await _model_db(prisma_client).create(data={**litellm_modeltable.json(exclude_none=True)})
        else:
            model_dict = await _model_db(prisma_client).upsert(
                where={"id": model_id},
                data={
                    "update": {**litellm_modeltable.json(exclude_none=True)},
                    "create": {**litellm_modeltable.json(exclude_none=True)},
                },
            )

        _model_id = model_dict.id

    return _model_id


async def _auto_add_team_members_to_organization(
    team: LiteLLM_TeamTable,
    organization: LiteLLM_OrganizationTableWithMembers,
    prisma_client: PrismaClient,
) -> None:
    """
    When moving a team to an org, ensure all team members are also org members.

    For SSO/Entra setups without SCIM, users join teams automatically on login but
    are never explicitly added to organizations. This silently upserts missing members
    rather than blocking the team move.
    """
    org_member_ids: Final = {m.user_id for m in organization.members} if organization.members else set()
    for member in team.members_with_roles:
        if member.user_id is None:
            continue
        if member.user_id == SpecialProxyStrings.default_user_id.value:
            continue
        if member.user_id in org_member_ids:
            continue
        if organization.organization_id is None:
            continue
        try:
            await add_member_to_organization(
                member=OrgMember(
                    user_id=member.user_id,
                    role=LitellmUserRoles.INTERNAL_USER,
                ),
                organization_id=organization.organization_id,
                prisma_client=prisma_client,
            )
        except Exception as e:
            verbose_proxy_logger.debug(
                "_auto_add_team_members_to_organization: skipping user_id=%s - %s",
                member.user_id,
                e,
            )


async def fetch_and_validate_organization(
    organization_id: str,
    existing_team_row: _ModelDumpRow,
    llm_router: Router | None,
    prisma_client: PrismaClient,
    user_api_key_dict: UserAPIKeyAuth | None = None,
) -> "prisma_models.LiteLLM_OrganizationTable":
    """
    Fetch and validate an organization for team update operations.

    Args:
        organization_id: The organization ID to fetch
        existing_team_row: The existing team row being updated
        llm_router: The LLM router instance
        prisma_client: The Prisma database client

    Returns:
        The organization row from the database

    Raises:
        HTTPException: If llm_router is None, organization not found, or validation fails
    """
    if llm_router is None:
        raise HTTPException(status_code=500, detail={"error": CommonProxyErrors.no_llm_router.value})

    organization_row: Final = await _org_db(prisma_client).find_unique(
        where={"organization_id": organization_id},
        include={"litellm_budget_table": True, "members": True, "teams": True},
    )

    if organization_row is None:
        raise HTTPException(
            status_code=404,
            detail={"error": f"Organization not found, passed organization_id={organization_id}"},
        )

    is_proxy_admin = user_api_key_dict is not None and user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN
    organization: Final = LiteLLM_OrganizationTableWithMembers.model_validate(organization_row.model_dump())
    validate_team_org_change(
        team=LiteLLM_TeamTable.model_validate(existing_team_row.model_dump()),
        organization=organization,
        llm_router=llm_router,
        is_proxy_admin=is_proxy_admin,
    )

    if is_proxy_admin:
        await _auto_add_team_members_to_organization(
            team=LiteLLM_TeamTable.model_validate(existing_team_row.model_dump()),
            organization=organization,
            prisma_client=prisma_client,
        )

    return organization_row


def validate_team_org_change(
    team: LiteLLM_TeamTable,
    organization: LiteLLM_OrganizationTableWithMembers,
    llm_router: Router,
    is_proxy_admin: bool = False,
) -> bool:
    """
    Validate that a team can be moved to an organization.

    - The org must have access to the team's models
    - The team budget cannot be greater than the org max_budget
    - For non-proxy-admins: all team members must already be org members
    - The team's tpm/rpm limit must be less than the org's tpm/rpm limit

    Proxy admins bypass the membership check and instead trigger auto-add of
    missing members (handled by the caller). This supports SSO/Entra setups
    where org membership tables are empty but proxy admins still need to group
    teams under orgs for budget/model governance.
    """

    # If the team's organization is the same as the new organization, return True
    # Since no changes are being made
    if team.organization_id == organization.organization_id:
        return True

    # Check if the org has access to the team's models
    if len(organization.models) > 0:
        if SpecialModelNames.all_proxy_models.value in organization.models:
            pass
        elif team.models is None or len(team.models) == 0:
            raise HTTPException(
                status_code=403,
                detail={
                    "error": "Cannot move team to organization. Team has access to all proxy models, but the organization does not."
                },
            )
        else:
            for model in team.models:
                can_org_access_model(
                    model=model,
                    org_object=organization,
                    llm_router=llm_router,
                )

    # Check if the team's budget is less than the org's max_budget
    if (
        team.max_budget
        and organization.litellm_budget_table
        and organization.litellm_budget_table.max_budget
        and team.max_budget > organization.litellm_budget_table.max_budget
    ):
        raise HTTPException(
            status_code=403,
            detail={
                "error": f"Cannot move team to organization. Team has max_budget {team.max_budget} that is greater than the organization's max_budget {organization.litellm_budget_table.max_budget}."
            },
        )

    # For non-proxy-admins, require all team members to already be org members.
    # This prevents a team admin from moving their team into an arbitrary org and
    # thereby injecting members into that org without org admin approval.
    if not is_proxy_admin:
        team_members: Final = [m.user_id for m in team.members_with_roles]
        org_members: Final = [m.user_id for m in organization.members] if organization.members else []
        not_in_org: Final = [
            m for m in team_members if m not in org_members and m != SpecialProxyStrings.default_user_id.value
        ]
        if len(not_in_org) > 0:
            raise HTTPException(
                status_code=403,
                detail={
                    "error": f"Cannot move team to organization. Team has user_id {not_in_org} that is not a member of the organization."
                },
            )

    # Check if the team's tpm/rpm limit is less than the org's tpm/rpm limit
    if (
        team.tpm_limit
        and organization.litellm_budget_table
        and organization.litellm_budget_table.tpm_limit
        and team.tpm_limit > organization.litellm_budget_table.tpm_limit
    ):
        raise HTTPException(
            status_code=403,
            detail={
                "error": f"Cannot move team to organization. Team has tpm_limit {team.tpm_limit} that is greater than the organization's tpm_limit {organization.litellm_budget_table.tpm_limit}."
            },
        )
    if (
        team.rpm_limit
        and organization.litellm_budget_table
        and organization.litellm_budget_table.rpm_limit
        and team.rpm_limit > organization.litellm_budget_table.rpm_limit
    ):
        raise HTTPException(
            status_code=403,
            detail={
                "error": f"Cannot move team to organization. Team has rpm_limit {team.rpm_limit} that is greater than the organization's rpm_limit {organization.litellm_budget_table.rpm_limit}."
            },
        )
    return True


@router.post("/team/update", tags=["team management"], dependencies=[Depends(user_api_key_auth)])
@management_endpoint_wrapper
async def update_team(
    data: UpdateTeamRequest,
    http_request: Request,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
    litellm_changed_by: str | None = Header(
        None,
        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",
    ),
):
    """
    Use `/team/member_add` AND `/team/member/delete` to add/remove new team members

    You can now update team budget / rate limits via /team/update

    Parameters:
    - team_id: str - The team id of the user. Required param.
    - team_alias: Optional[str] - User defined team alias
    - team_member_permissions: Optional[List[str]] - A list of routes that non-admin team members can access. example: ["/key/generate", "/key/update", "/key/delete"]
    - metadata: Optional[dict] - Metadata for team, store information for team. Example metadata = {"team": "core-infra", "app": "app2", "email": "ishaan@berri.ai" }
    - tpm_limit: Optional[int] - The TPM (Tokens Per Minute) limit for this team - all keys with this team_id will have at max this TPM limit
    - rpm_limit: Optional[int] - The RPM (Requests Per Minute) limit for this team - all keys associated with this team_id will have at max this RPM limit
    - max_budget: Optional[float] - The maximum budget allocated to the team - all keys for this team_id will have at max this max_budget
    - soft_budget: Optional[float] - The soft budget threshold for the team. If max_budget is set (either in the request or existing), soft_budget must be strictly lower than max_budget. Can be set independently if max_budget is not set.
    - budget_duration: Optional[str] - The duration of the budget for the team. Doc [here](https://docs.litellm.ai/docs/proxy/team_budgets)
    - models: Optional[list] - A list of models associated with the team - all keys for this team_id will have at most, these models. If empty, assumes all models are allowed.
    - prompts: Optional[List[str]] - List of prompts that the team is allowed to use.
    - blocked: bool - Flag indicating if the team is blocked or not - will stop all calls from keys with this team_id.
    - tags: Optional[List[str]] - Tags for [tracking spend](https://litellm.vercel.app/docs/proxy/enterprise#tracking-spend-for-custom-tags) and/or doing [tag-based routing](https://litellm.vercel.app/docs/proxy/tag_routing).
    - organization_id: Optional[str] - The organization id of the team. Default is None. Create via `/organization/new`.
    - model_aliases: Optional[dict] - Model aliases for the team. [Docs](https://docs.litellm.ai/docs/proxy/team_based_routing#create-team-with-model-alias)
    - guardrails: Optional[List[str]] - Guardrails for the team. [Docs](https://docs.litellm.ai/docs/proxy/guardrails)
    - policies: Optional[List[str]] - Policies for the team. [Docs](https://docs.litellm.ai/docs/proxy/guardrails/guardrail_policies)
    - disable_global_guardrails: Optional[bool] - Whether to disable global guardrails for the key.
    - object_permission: Optional[LiteLLM_ObjectPermissionBase] - team-specific object permission. Example - {"vector_stores": ["vector_store_1", "vector_store_2"], "agents": ["agent_1", "agent_2"], "agent_access_groups": ["dev_group"]}. IF null or {} then no object permission.
    - team_member_budget: Optional[float] - The maximum budget allocated to an individual team member.
    - team_member_budget_duration: Optional[str] - The duration of the budget for the team member. Doc [here](https://docs.litellm.ai/docs/proxy/team_budgets)
    - team_member_rpm_limit: Optional[int] - The RPM (Requests Per Minute) limit for individual team members.
    - team_member_tpm_limit: Optional[int] - The TPM (Tokens Per Minute) limit for individual team members.
    - team_member_key_duration: Optional[str] - The duration for a team member's key. e.g. "1d", "1w", "1mo"
    - allowed_passthrough_routes: Optional[List[str]] - List of allowed pass through routes for the team.
    - model_rpm_limit: Optional[Dict[str, int]] - The RPM (Requests Per Minute) limit per model for this team. Example: {"gpt-4": 100, "gpt-3.5-turbo": 200}
    - model_tpm_limit: Optional[Dict[str, int]] - The TPM (Tokens Per Minute) limit per model for this team. Example: {"gpt-4": 10000, "gpt-3.5-turbo": 20000}
    - default_estimated_output_tokens: Optional[int] - Expected output tokens reserved for TPM limiting when a request omits max_tokens, for keys on this team that do not set their own. Positive integer.
    - default_estimated_output_tokens_per_model: Optional[Dict[str, int]] - Per-model override of the above. Example: {"gpt-4": 4096, "gpt-3.5-turbo": 1024}
    - mcp_rpm_limit: Optional[Dict[str, int]] - Per-MCP-server RPM limit for this team, keyed by MCP server name (alias if set, else the configured name). Example: {"github": 100, "slack": 200}. Applied across all keys for this team.
    Example - update team TPM Limit
    - allowed_vector_store_indexes: Optional[List[dict]] - List of allowed vector store indexes for the key. Example - [{"index_name": "my-index", "index_permissions": ["write", "read"]}]. If specified, the key will only be able to use these specific vector store indexes. Create index, using `/v1/indexes` endpoint.
    - secret_manager_settings: Optional[dict] - Secret manager settings for the team. [Docs](https://docs.litellm.ai/docs/secret_managers/overview)
    - router_settings: Optional[UpdateRouterConfig] - team-specific router settings. Example - {"model_group_retry_policy": {"gpt-4": {"RateLimitErrorRetries": 5}}}. IF null or {} then no router settings.
    - access_group_ids: Optional[List[str]] - List of access group IDs to associate with the team. Access groups define which models the team can access. Example - ["access_group_1", "access_group_2"].
    - enforced_file_expires_after: Optional[dict] - Enforced file expiration policy for the team. Keys created under this team will inherit this policy for file uploads. Example - {"anchor": "created_at", "days": 30}.
    - enforced_batch_output_expires_after: Optional[dict] - Enforced batch output file expiration policy for the team. Keys created under this team will inherit this policy for batch output files. Example - {"anchor": "created_at", "days": 30}.
    - budget_limits: Optional[list] - List of concurrent budget windows for the team. 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"}].
    - default_team_member_models: Optional[List[str]] - Default models assigned to new team members when they join this team. Must be a subset of the team's models.

    ```
    curl --location 'http://0.0.0.0:4000/team/update' \
    --header 'Authorization: Bearer sk-1234' \
    --header 'Content-Type: application/json' \
    --data-raw '{
        "team_id": "8d916b1c-510d-4894-a334-1c16a93344f5",
        "tpm_limit": 100
    }'
    ```

    Example - Update Team `max_budget` budget
    ```
    curl --location 'http://0.0.0.0:4000/team/update' \
    --header 'Authorization: Bearer sk-1234' \
    --header 'Content-Type: application/json' \
    --data-raw '{
        "team_id": "8d916b1c-510d-4894-a334-1c16a93344f5",
        "max_budget": 10
    }'
    ```
    """
    try:
        from litellm.proxy.management_helpers.audit_logs import is_audit_logging_enabled
        from litellm.proxy.proxy_server import (
            litellm_proxy_admin_name,
            llm_router,
            prisma_client,
            proxy_logging_obj,
            user_api_key_cache,
        )

        if prisma_client is None:
            raise HTTPException(
                status_code=500,
                detail={"error": CommonProxyErrors.db_not_connected_error.value},
            )

        if data.team_id is None:
            raise HTTPException(status_code=400, detail={"error": "No team id passed in"})
        verbose_proxy_logger.debug("/team/update - %s", data)

        # Validate budget values are not negative
        if data.max_budget is not None and (not math.isfinite(data.max_budget) or data.max_budget < 0):
            raise HTTPException(
                status_code=400,
                detail={"error": f"max_budget must be a non-negative finite number. Received: {data.max_budget}"},
            )
        if data.team_member_budget is not None and (
            not math.isfinite(data.team_member_budget) or data.team_member_budget < 0
        ):
            raise HTTPException(
                status_code=400,
                detail={
                    "error": f"team_member_budget must be a non-negative finite number. Received: {data.team_member_budget}"
                },
            )
        if data.soft_budget is not None and (not math.isfinite(data.soft_budget) or data.soft_budget < 0):
            raise HTTPException(
                status_code=400,
                detail={"error": f"soft_budget must be a non-negative finite number. Received: {data.soft_budget}"},
            )

        validate_budget_duration(data.budget_duration)
        validate_budget_duration(data.team_member_budget_duration)

        existing_team_row = await _raw_team_db(TeamRepository(prisma_client)).find_unique(
            where={"team_id": data.team_id}
        )

        if existing_team_row is None:
            raise HTTPException(
                status_code=404,
                detail={"error": f"Team not found, passed team_id={data.team_id}"},
            )

        # Verify caller has access to manage this team
        await _verify_team_access(
            team_obj=LiteLLM_TeamTable.model_validate(existing_team_row.model_dump()),
            user_api_key_dict=user_api_key_dict,
        )

        _existing_team_metadata: Final[object] = getattr(existing_team_row, "metadata", None)
        enforce_output_token_estimates_are_admin_only(
            data=data,
            existing_metadata=_existing_team_metadata if isinstance(_existing_team_metadata, dict) else None,
            user_api_key_dict=user_api_key_dict,
            entity="team",
        )
        enforce_batch_enqueued_token_limit_is_admin_only(
            data=data,
            existing_metadata=_existing_team_metadata if isinstance(_existing_team_metadata, dict) else None,
            user_api_key_dict=user_api_key_dict,
            entity="team",
        )

        _check_passthrough_routes_caller_permission(data, user_api_key_dict, entity="team")

        if data.soft_budget is not None:
            max_budget_to_check = data.max_budget if data.max_budget is not None else existing_team_row.max_budget
            if max_budget_to_check is not None:
                if data.soft_budget >= max_budget_to_check:
                    raise HTTPException(
                        status_code=400,
                        detail={
                            "error": f"soft_budget ({data.soft_budget}) must be strictly lower than max_budget ({max_budget_to_check})"
                        },
                    )

        if data.max_budget is not None:
            existing_soft_budget: Final[object] = _as_object(getattr(existing_team_row, "soft_budget", None))
            soft_budget_to_check: Final = data.soft_budget if data.soft_budget is not None else existing_soft_budget
            if soft_budget_to_check is not None and isinstance(soft_budget_to_check, (int, float)):
                if data.max_budget <= soft_budget_to_check:
                    raise HTTPException(
                        status_code=400,
                        detail={
                            "error": f"max_budget ({data.max_budget}) must be strictly greater than soft_budget ({soft_budget_to_check})"
                        },
                    )

        if data.organization_id is not None and len(data.organization_id) > 0:  # allow unsetting the organization_id
            # If the caller is relocating the team to a different org, they
            # must also be PROXY_ADMIN or an org-admin of the DESTINATION org.
            # _verify_team_access above only checked the team's CURRENT org,
            # so without this gate an org-admin could hand their team to any
            # other org (or capture a team from another org they once
            # administered into a new destination).
            current_org_id: Final = getattr(existing_team_row, "organization_id", None)
            if (
                data.organization_id != current_org_id
                and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value
            ):
                # Is the caller org_admin of the destination org?
                caller_memberships: Final = (
                    await _org_membership_db(prisma_client).find_many(
                        where={
                            "user_id": user_api_key_dict.user_id,
                            "organization_id": data.organization_id,
                            "user_role": LitellmUserRoles.ORG_ADMIN.value,
                        }
                    )
                    if user_api_key_dict.user_id
                    else []
                )
                if not caller_memberships:
                    raise HTTPException(
                        status_code=403,
                        detail={
                            "error": (
                                "Relocating a team to a different organization "
                                "requires PROXY_ADMIN or org-admin of the "
                                "destination org."
                            )
                        },
                    )

            await fetch_and_validate_organization(
                organization_id=data.organization_id,
                existing_team_row=existing_team_row,
                llm_router=llm_router,
                prisma_client=prisma_client,
                user_api_key_dict=user_api_key_dict,
            )
        elif data.organization_id is not None and len(data.organization_id) == 0:
            # unsetting the organization_id
            data.organization_id = None

        # check org team limits - if updating team that belongs to an org
        org_id_to_check: Final[object] = _as_object(
            data.organization_id if data.organization_id is not None else existing_team_row.organization_id
        )
        if org_id_to_check is not None and isinstance(org_id_to_check, str) and prisma_client is not None:
            org_table: Final = await get_org_object(
                org_id=org_id_to_check,
                user_api_key_cache=user_api_key_cache,
                prisma_client=prisma_client,
            )
            if org_table is not None:
                await _check_org_team_limits(
                    org_table=org_table,
                    data=data,
                    prisma_client=prisma_client,
                )

        # Only a proxy admin may grow a standalone team's spend ceiling.
        # Org-scoped teams are validated by _check_org_team_limits() above.
        if org_id_to_check is None:
            _check_team_budget_update_authority(
                data=data,
                user_api_key_dict=user_api_key_dict,
                existing_team_max_budget=existing_team_row.max_budget,
            )

        updated_kv = data.json(exclude_unset=True)

        # Drop server-owned metadata keys from caller input so they can only
        # be written by the same code path that creates the underlying rows.
        if isinstance(updated_kv.get("metadata"), dict):
            TeamMemberBudgetHandler.strip_system_managed_metadata_keys(updated_kv["metadata"])

        if "metadata" in updated_kv:
            stored_metadata: Final[Mapping[str, JsonValue] | None] = (
                {  # mutable-ok: the validator payload's isinstance guard requires a plain dict
                    key: value
                    for key, value in existing_team_row.metadata.items()
                    if key not in TeamMemberBudgetHandler.SYSTEM_MANAGED_METADATA_KEYS
                }
                if isinstance(existing_team_row.metadata, dict)
                else None
            )
            await validate_team_metadata_if_configured(
                operation="update",
                metadata=updated_kv.get("metadata"),
                existing_metadata=stored_metadata,
                team_id=data.team_id,
                team_alias=data.team_alias if data.team_alias is not None else existing_team_row.team_alias,
                user_api_key_dict=user_api_key_dict,
            )

        # Check budget_duration and budget_reset_at
        _set_budget_reset_at(data, updated_kv)

        _team_member_fields_in_request: Final = {
            field
            for field in [
                "team_member_budget",
                "team_member_rpm_limit",
                "team_member_tpm_limit",
                "team_member_budget_duration",
            ]
            if field in updated_kv
        }

        if _team_member_fields_in_request and TeamMemberBudgetHandler.should_create_budget(
            team_member_budget=data.team_member_budget,
            team_member_rpm_limit=data.team_member_rpm_limit,
            team_member_tpm_limit=data.team_member_tpm_limit,
            team_member_budget_duration=data.team_member_budget_duration,
        ):
            updated_kv = await TeamMemberBudgetHandler.upsert_team_member_budget_table(
                team_table=existing_team_row,
                user_api_key_dict=user_api_key_dict,
                updated_kv=updated_kv,
                team_member_budget=data.team_member_budget,
                team_member_rpm_limit=data.team_member_rpm_limit,
                team_member_tpm_limit=data.team_member_tpm_limit,
                team_member_budget_duration=data.team_member_budget_duration,
                explicitly_set_fields=_team_member_fields_in_request,
            )
            # Backfill team_memberships for members who joined before the
            # budget was configured — they won't have a membership row yet.
            _backfill_budget_id: Final = (updated_kv.get("metadata") or {}).get("team_member_budget_id")
            if _backfill_budget_id and existing_team_row.members_with_roles:
                await TeamMemberBudgetHandler.backfill_team_member_budget_entries(
                    team_id=data.team_id,
                    members_with_roles=existing_team_row.members_with_roles,
                    team_member_budget_id=_backfill_budget_id,
                    prisma_client=prisma_client,
                )
        elif _team_member_fields_in_request:
            updated_kv = await TeamMemberBudgetHandler.clear_team_member_budget_fields(
                team_table=existing_team_row,
                user_api_key_dict=user_api_key_dict,
                updated_kv=updated_kv,
                explicitly_set_fields=_team_member_fields_in_request,
            )
        else:
            TeamMemberBudgetHandler._clean_team_member_fields(updated_kv)

        # Check object permission
        if data.object_permission is not None:
            await enforce_all_proxy_mcp_servers_grant_is_admin_only(
                requested_mcp_servers=data.object_permission.mcp_servers,
                existing_object_permission_id=existing_team_row.object_permission_id,
                is_proxy_admin=user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN,
                prisma_client=prisma_client,
            )
            updated_kv = await handle_update_object_permission(
                data_json=updated_kv,
                existing_team_row=existing_team_row,
            )

        # update team metadata fields
        _update_metadata_fields(updated_kv=updated_kv)

        if updated_kv.get("metadata") is not None:
            updated_kv["metadata"] = encrypt_callback_vars(updated_kv["metadata"])

        if "model_aliases" in updated_kv:
            updated_kv.pop("model_aliases")
            _model_id: Final = await _update_model_table(
                data=data,
                model_id=existing_team_row.model_id,
                prisma_client=prisma_client,
                user_api_key_dict=user_api_key_dict,
                litellm_proxy_admin_name=litellm_proxy_admin_name,
            )
            if _model_id is not None:
                updated_kv["model_id"] = _model_id

        # Serialize router_settings to JSON if present (matching key update pattern)
        if "router_settings" in updated_kv and updated_kv["router_settings"] is not None:
            updated_kv["router_settings"] = safe_dumps(updated_kv["router_settings"])

        updated_kv = prisma_client.jsonify_team_object(db_data=updated_kv)
        team_update_data: Final[Mapping[str, object]] = updated_kv
        team_row: Final = await _team_db(prisma_client).update(
            where={"team_id": data.team_id},
            data=team_update_data,
            # `object_permission` is included so `_refresh_cached_team`
            # doesn't write a cached team with the relation nulled out.
            # See team_model_add for the full rationale.
            include={
                "litellm_model_table": True,
                "object_permission": True,
            },
        )

        if team_row is None or team_row.team_id is None:
            raise HTTPException(
                status_code=400,
                detail={"error": f"Team doesn't exist. Got={team_row}"},
            )

        verbose_proxy_logger.info("Successfully updated team - %s, info", team_row.team_id)
        await sync_team_access_group_membership(prisma_client=prisma_client, team_id=team_row.team_id)
        await _refresh_cached_team(
            team_row=team_row,
            user_api_key_cache=user_api_key_cache,
            proxy_logging_obj=proxy_logging_obj,
        )

        if is_audit_logging_enabled():
            await _create_team_update_audit_log(
                existing_team_row=existing_team_row,
                updated_kv=updated_kv,
                team_id=data.team_id,
                litellm_changed_by=litellm_changed_by,
                user_api_key_dict=user_api_key_dict,
                litellm_proxy_admin_name=litellm_proxy_admin_name,
            )

        return {"team_id": team_row.team_id, "data": team_row}
    except Exception as e:
        raise handle_exception_on_proxy(e)


@router.patch(
    "/team/{team_id}",
    tags=["team management"],
    dependencies=[Depends(user_api_key_auth)],
    response_model=LiteLLM_TeamTable,
)
async def patch_team(
    team_id: str,
    data: PatchTeamRequest,
    http_request: Request,
    user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
    litellm_changed_by: Annotated[
        str | None,
        Header(
            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",
        ),
    ] = None,
):
    """
    Partially update a team using RFC 7386 JSON Merge Patch semantics.

    `team_id` is taken from the path; a `team_id` in the body is accepted only when it
    matches. `metadata` is merged with the team's stored metadata rather than replacing
    it: an omitted key is preserved, `key: null` deletes it, and any other value
    overwrites (recursing into nested objects). Every other field behaves exactly like
    `POST /team/update` (omitted preserves, a value overwrites). Returns the full
    updated team.

    ```
    curl --location --request PATCH 'http://0.0.0.0:4000/team/8d916b1c-510d-4894-a334-1c16a93344f5' \
    --header 'Authorization: Bearer sk-1234' \
    --header 'Content-Type: application/json' \
    --data-raw '{
        "metadata": {"cost_center": "1234", "deprecated_key": null}
    }'
    ```
    """
    from litellm.proxy.proxy_server import prisma_client

    try:
        if prisma_client is None:
            raise HTTPException(
                status_code=500,
                detail={"error": CommonProxyErrors.db_not_connected_error.value},
            )

        if data.team_id is not None and data.team_id != team_id:
            raise HTTPException(
                status_code=400,
                detail={"error": f"team_id in body ({data.team_id}) does not match team_id in path ({team_id})"},
            )

        patch_fields: Final = data.model_dump(exclude_unset=True, exclude={"team_id"})

        if "metadata" in patch_fields:
            existing_team_row: Final = await _team_db(prisma_client).find_unique(where={"team_id": team_id})
            if existing_team_row is None:
                raise HTTPException(
                    status_code=404,
                    detail={"error": f"Team not found, passed team_id={team_id}"},
                )
            existing_metadata = existing_team_row.metadata if isinstance(existing_team_row.metadata, dict) else {}
            patch_fields["metadata"] = apply_json_merge_patch(existing_metadata, patch_fields["metadata"])

        update_request: Final = UpdateTeamRequest.model_validate({"team_id": team_id, **patch_fields})

        result: Final = await update_team(
            data=update_request,
            http_request=http_request,
            user_api_key_dict=user_api_key_dict,
            litellm_changed_by=litellm_changed_by,
        )
        return result["data"]
    except Exception as e:  # noqa: BLE001  # normalize every failure to the proxy exception contract
        raise handle_exception_on_proxy(e)


def _set_budget_reset_at(data: UpdateTeamRequest, updated_kv: dict) -> None:
    """Set budget_reset_at in updated_kv if budget_duration is provided."""
    if data.budget_duration is not None:
        from litellm.proxy.common_utils.timezone_utils import get_budget_reset_time

        reset_at: Final = get_budget_reset_time(budget_duration=data.budget_duration)
        updated_kv["budget_reset_at"] = reset_at
    elif "budget_duration" in updated_kv and updated_kv["budget_duration"] is None:
        updated_kv["budget_reset_at"] = None

    if data.budget_limits is not None and len(data.budget_limits) > 0:
        from litellm.proxy.common_utils.timezone_utils import get_budget_reset_time

        initialized_windows: Final = []
        for window in data.budget_limits:
            w = window if isinstance(window, dict) else window.model_dump()
            w["reset_at"] = get_budget_reset_time(budget_duration=w["budget_duration"]).isoformat()
            initialized_windows.append(w)
        updated_kv["budget_limits"] = json.dumps(initialized_windows)


async def handle_update_object_permission(data_json: dict, existing_team_row: _ObjectPermissionRow) -> dict:
    """
    Handle the update of object permission for a team.

    - IF there's no object_permission_id, then create a new entry in LiteLLM_ObjectPermissionTable
    - IF there's an object_permission_id, then update the entry in LiteLLM_ObjectPermissionTable
    """
    from litellm.proxy.proxy_server import prisma_client

    # Use the common helper to handle the object permission update
    object_permission_id: Final = await handle_update_object_permission_common(
        data_json=data_json,
        existing_object_permission_id=existing_team_row.object_permission_id,
        prisma_client=prisma_client,
    )

    # Add the object_permission_id to data_json if one was created/updated
    if object_permission_id is not None:
        data_json["object_permission_id"] = object_permission_id
        verbose_proxy_logger.debug("updated object_permission_id: %s", object_permission_id)

    return data_json


def _check_team_member_admin_add(
    member: Member | list[Member],
    premium_user: bool,
):
    if isinstance(member, Member) and member.role == "admin":
        if premium_user is not True:
            raise ValueError(f"Assigning team admins is a premium feature. {CommonProxyErrors.not_premium_user.value}")
    elif isinstance(member, list):
        for m in member:
            if m.role == "admin":
                if premium_user is not True:
                    raise ValueError(
                        f"Assigning team admins is a premium feature. Got={m}. {CommonProxyErrors.not_premium_user.value}. "
                    )


def team_call_validation_checks(
    prisma_client: PrismaClient | None,
    data: TeamMemberAddRequest,
    premium_user: bool,
):
    if prisma_client is None:
        raise HTTPException(status_code=500, detail={"error": "No db connected"})

    if data.team_id is None:
        raise HTTPException(status_code=400, detail={"error": "No team id passed in"})

    if data.member is None:
        raise HTTPException(status_code=400, detail={"error": "No member/members passed in"})

    try:
        _check_team_member_admin_add(
            member=data.member,
            premium_user=premium_user,
        )
    except Exception as e:
        raise HTTPException(status_code=400, detail={"error": str(e)})


def team_member_add_duplication_check(
    data: TeamMemberAddRequest,
    existing_team_row: LiteLLM_TeamTable,
):
    """
    Check if a member already exists in the team.
    This check is done BEFORE we create/fetch the user, so it only prevents
    obvious duplicates where both user_id and user_email match exactly.
    """

    invalid_team_members: Final = []

    def _check_member_duplication(member: Member):
        if member.user_id is not None:
            for existing_member in existing_team_row.members_with_roles:
                if existing_member.user_id == member.user_id:
                    invalid_team_members.append(member)

        # Check by user_email if provided
        if member.user_email is not None:
            for existing_member in existing_team_row.members_with_roles:
                if existing_member.user_email == member.user_email:
                    invalid_team_members.append(member)

    # First, populate the invalid_team_members list by checking for duplicates
    if isinstance(data.member, Member):
        _check_member_duplication(data.member)
    elif isinstance(data.member, list):
        for m in data.member:
            _check_member_duplication(m)

    # Then check the populated list and raise exceptions if needed
    if isinstance(data.member, list) and len(invalid_team_members) == len(data.member):
        raise ProxyException(
            message=f"All users are already in team. Existing members={existing_team_row.members_with_roles}",
            type=ProxyErrorTypes.team_member_already_in_team,
            param="member",
            code="400",
        )
    elif isinstance(data.member, Member) and len(invalid_team_members) == 1:
        raise ProxyException(
            message=f"User already in team. Member: user_id={data.member.user_id}, user_email={data.member.user_email}. Existing members={existing_team_row.members_with_roles}",
            type=ProxyErrorTypes.team_member_already_in_team,
            param="member",
            code="400",
        )
    elif len(invalid_team_members) > 0:
        verbose_proxy_logger.info(
            "Some users are already in team. Existing members=%s. Duplicate members=%s",
            existing_team_row.members_with_roles,
            invalid_team_members,
        )


async def _validate_team_member_add_permissions(
    user_api_key_dict: UserAPIKeyAuth,
    complete_team_data: LiteLLM_TeamTable,
    data: TeamMemberAddRequest,
) -> None:
    """Validate if user has permission to add members to the team.

    Standard users can self-join an *available team*, but the bypass
    must not be allowed to escalate them to ``role=admin`` or to add
    other users into the team.  When access is granted via the
    available-team bypass we therefore enforce that every member in
    the request matches the caller's own ``user_id`` and is being
    added with ``role="user"``.
    """
    if getattr(user_api_key_dict, "user_role", None) == LitellmUserRoles.PROXY_ADMIN.value:
        return
    if _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=complete_team_data):
        return
    if await _is_user_org_admin_for_team(user_api_key_dict=user_api_key_dict, team_obj=complete_team_data):
        return

    if not _is_available_team(
        team_id=complete_team_data.team_id,
        user_api_key_dict=user_api_key_dict,
    ):
        raise HTTPException(
            status_code=403,
            detail={
                "error": "Call not allowed. User not proxy admin OR team admin. route={}, team_id={}".format(
                    "/team/member_add",
                    complete_team_data.team_id,
                )
            },
        )

    # Available-team self-join grants only the ability to join; per-member
    # budget and model controls stay admin-only.  Reject them here so a
    # self-joining non-admin cannot set their own cap, reset window, or model
    # scope via the bypass.
    if data.max_budget_in_team is not None or data.budget_duration is not None or data.allowed_models is not None:
        raise HTTPException(
            status_code=403,
            detail={
                "error": (
                    "Available-team self-join cannot set per-member budget or "
                    "model controls (max_budget_in_team, budget_duration, "
                    "allowed_models); these are admin-only."
                )
            },
        )

    # Available-team self-join: caller may add only themselves, only as a
    # standard user.  Enforce that here so the bypass cannot be used as a
    # privilege-escalation or cross-user-injection primitive.
    members: Final = data.member if isinstance(data.member, list) else [data.member]
    caller_user_id: Final = getattr(user_api_key_dict, "user_id", None)
    for member in members:
        if getattr(member, "role", "user") != "user":
            raise HTTPException(
                status_code=403,
                detail={
                    "error": (
                        "Available-team self-join cannot assign 'admin' role. "
                        "Only proxy/team/org admins can add admins to a team."
                    )
                },
            )
        member_user_id = getattr(member, "user_id", None)
        if not caller_user_id or not member_user_id or member_user_id != caller_user_id:
            raise HTTPException(
                status_code=403,
                detail={
                    "error": (
                        "Available-team self-join can only add the caller "
                        "(user_id must match the authenticated user's user_id)."
                    )
                },
            )


async def _process_team_members(
    data: TeamMemberAddRequest,
    complete_team_data: LiteLLM_TeamTable,
    prisma_client: PrismaClient,
    user_api_key_dict: UserAPIKeyAuth,
    litellm_proxy_admin_name: str,
    tx: MemberWriteTx | None = None,
) -> tuple[list[LiteLLM_UserTable], list[LiteLLM_TeamMembership]]:
    """Process and add new team members.

    ``tx`` is the caller's open transaction, when it has one, so the member writes run on the
    connection it already holds instead of checking out a second one.
    """
    updated_users: Final[list[LiteLLM_UserTable]] = []
    updated_team_memberships: Final[list[LiteLLM_TeamMembership]] = []

    default_team_budget_id: Final = (
        complete_team_data.metadata.get("team_member_budget_id") if complete_team_data.metadata is not None else None
    )

    # Resolve allowed_models: explicit request value, or fall back to team's default_team_member_models
    member_allowed_models = data.allowed_models
    team_default_member_models: Final = getattr(complete_team_data, "default_team_member_models", None)
    if member_allowed_models is None and team_default_member_models:
        member_allowed_models = team_default_member_models

    if isinstance(data.member, Member):
        try:
            updated_user, updated_tm = await add_new_member(
                new_member=data.member,
                max_budget_in_team=data.max_budget_in_team,
                prisma_client=prisma_client,
                user_api_key_dict=user_api_key_dict,
                litellm_proxy_admin_name=litellm_proxy_admin_name,
                team_id=data.team_id,
                default_team_budget_id=default_team_budget_id,
                allowed_models=member_allowed_models,
                budget_duration=data.budget_duration,
                tx=tx,
            )
        except Exception as e:
            raise HTTPException(
                status_code=500,
                detail={"error": f"Unable to add user - {data.member}, to team - {data.team_id}, for reason - {e}"},
            )
        updated_users.append(updated_user)
        if updated_tm is not None:
            updated_team_memberships.append(updated_tm)
    elif isinstance(data.member, list):
        for m in data.member:
            try:
                updated_user, updated_tm = await add_new_member(
                    new_member=m,
                    max_budget_in_team=data.max_budget_in_team,
                    prisma_client=prisma_client,
                    user_api_key_dict=user_api_key_dict,
                    litellm_proxy_admin_name=litellm_proxy_admin_name,
                    team_id=data.team_id,
                    default_team_budget_id=default_team_budget_id,
                    allowed_models=member_allowed_models,
                    budget_duration=data.budget_duration,
                    tx=tx,
                )
            except Exception as e:
                raise HTTPException(
                    status_code=500,
                    detail={"error": f"Unable to add user - {m}, to team - {data.team_id}, for reason - {e}"},
                )
            updated_users.append(updated_user)
            if updated_tm is not None:
                updated_team_memberships.append(updated_tm)

    return updated_users, updated_team_memberships


def _resolve_member_identity(member: Member, updated_users: Sequence[LiteLLM_UserTable]) -> Member:
    """Return ``member`` with whichever of ``user_id`` / ``user_email`` the caller left out filled in.

    The roster entry is a snapshot, so whatever is missing here is missing for good.
    Resolution runs both ways off the user rows the add just touched: added by email
    -> stamp the user_id, added by user_id -> stamp the email. A value the caller
    supplied is never overwritten.
    """
    resolved_user_id: Final = member.user_id or next(
        (
            user.user_id
            for user in updated_users
            if member.user_email is not None and user.user_email == member.user_email
        ),
        None,
    )
    resolved_user_email: Final = member.user_email or next(
        (
            user.user_email
            for user in updated_users
            if resolved_user_id is not None and user.user_id == resolved_user_id and user.user_email is not None
        ),
        None,
    )
    return member.model_copy(
        update={  # mutable-ok: pydantic update payload
            "user_id": resolved_user_id,
            "user_email": resolved_user_email,
        }
    )


def _member_already_in_team(member: Member, complete_team_data: LiteLLM_TeamTable) -> bool:
    return any(
        (member.user_id is not None and existing_member.user_id == member.user_id)
        or (member.user_email is not None and existing_member.user_email == member.user_email)
        for existing_member in complete_team_data.members_with_roles
    )


async def _update_team_members_list(
    data: TeamMemberAddRequest,
    complete_team_data: LiteLLM_TeamTable,
    updated_users: list[LiteLLM_UserTable],
) -> None:
    """Update the team's members_with_roles list."""
    requested_members: Final[Sequence[Member]] = (
        (data.member,) if isinstance(data.member, Member) else tuple(data.member)
    )
    resolved_members: Final = tuple(_resolve_member_identity(m, updated_users) for m in requested_members)

    # extend() consumes the generator as it appends, so a member already added by this
    # same call is seen by the next _member_already_in_team check - the batch dedupes
    # against itself exactly as the append-one-at-a-time loop this replaced did.
    complete_team_data.members_with_roles.extend(  # rebind-ok: this helper's contract is to grow the caller's roster in place
        m for m in resolved_members if not _member_already_in_team(m, complete_team_data)
    )


async def _add_team_members_to_team(
    data: TeamMemberAddRequest,
    complete_team_data: LiteLLM_TeamTable,
    prisma_client: PrismaClient,
    user_api_key_dict: UserAPIKeyAuth,
    litellm_proxy_admin_name: str,
) -> tuple["prisma_models.LiteLLM_TeamTable", list[LiteLLM_UserTable], list[LiteLLM_TeamMembership]]:
    """Add team members to the team, under the team's advisory lock.

    The lock (``TEAM_ADVISORY_LOCK_SQL``, keyed on the team id) is taken first, and the
    team is re-read under it before any write, so a delete that already committed is
    visible here before this call writes anything: the user and membership writes only
    happen once the re-read proves the team is still live. /team/delete takes the same
    lock around its own sweep-and-delete, so the two can never interleave; whichever
    acquires the lock first runs to completion before the other's re-read can proceed.

    The user and membership writes run on this transaction too, not on a second
    connection from the pool: a lock waiter that needs a connection it hasn't got yet is
    a waiter that can deadlock the pool, since enough concurrent adds for one team would
    hold every connection waiting on the lock while the holder waits for a free one.
    """
    gone_detail: Final[_ErrorDetail] = {"error": f"Team={data.team_id} was deleted while this member add was running"}
    async with prisma_client.tx() as tx:
        await tx.query_raw(TEAM_ADVISORY_LOCK_SQL, data.team_id)

        locked_members: Final = await TeamRepository(prisma_client).get_members_with_roles_locked(tx, data.team_id)
        if locked_members is None:
            raise HTTPException(status_code=404, detail=gone_detail)
        complete_team_data.members_with_roles = locked_members

        updated_users, updated_team_memberships = await _process_team_members(
            data=data,
            complete_team_data=complete_team_data,
            prisma_client=prisma_client,
            user_api_key_dict=user_api_key_dict,
            litellm_proxy_admin_name=litellm_proxy_admin_name,
            tx=tx,
        )

        await _update_team_members_list(
            data=data,
            complete_team_data=complete_team_data,
            updated_users=updated_users,
        )

        _db_team_members: Final = [m.model_dump() for m in complete_team_data.members_with_roles]
        updated_team: Final = await _team_tx_db(tx).update(
            where={"team_id": data.team_id},
            data={"members_with_roles": json.dumps(_db_team_members)},
        )
        if updated_team is None:
            raise HTTPException(status_code=404, detail=gone_detail)

    return updated_team, updated_users, updated_team_memberships


def _emit_team_members_metric(team: LiteLLM_TeamTable) -> None:
    """Update the Prometheus team members gauge after a membership change.

    No-ops when the Prometheus callback is not registered, and never lets a
    metric failure break the team add/delete request.
    """
    prometheus_logger: Final = PrometheusLogger.get_instance()
    if prometheus_logger is None:
        return
    try:
        prometheus_logger.set_team_members_metric(team)
    except Exception as e:
        verbose_proxy_logger.debug("Prometheus: failed to emit team members metric: %s", str(e))


async def _resolve_existing_member_user_ids(
    members: Sequence[Member],
    prisma_client: PrismaClient,
) -> frozenset[str]:
    """Return the caller-supplied user_ids that already have a user row.

    Resolved with a single query so the number of members in the request does
    not translate into that many concurrent connections.
    """
    requested_user_ids: Final = frozenset(member.user_id for member in members if member.user_id is not None)
    if not requested_user_ids:
        return frozenset()

    found: Final = await _user_id_rows_db(UserRepository(prisma_client)).find_many(
        where={  # mutable-ok: Prisma query filters are dict-shaped
            "user_id": {  # mutable-ok: Prisma query filters are dict-shaped
                "in": sorted(requested_user_ids)
            }
        }
    )
    return frozenset(user.user_id for user in found or () if user.user_id is not None)


def _pre_existing_user_ids(
    members: Sequence[Member],
    caller_supplied_user_ids: frozenset[str],
    existing_user_ids: frozenset[str],
) -> frozenset[str]:
    """Return the user_ids that already had a user row before this request.

    Combines the caller-supplied ids that resolved to a user with the ids
    ``_validate_and_populate_member_user_info`` filled in, which it only does
    from a matched user row. Deriving it that way keeps this in step with the
    email matching that resolution performs, rather than repeating it here.
    """
    populated_user_ids: Final = frozenset(
        member.user_id
        for member in members
        if member.user_id is not None and member.user_id not in caller_supplied_user_ids
    )
    return existing_user_ids | populated_user_ids


_MAX_REPORTED_UNKNOWN_USER_IDS: Final = 10


def _validate_member_user_id_provisioning(
    members: Sequence[Member],
    existing_user_ids: frozenset[str],
    user_api_key_dict: UserAPIKeyAuth,
) -> None:
    """Restrict adding a caller-chosen user_id that has no user row yet to proxy admins.

    Team and org admins keep the ability to add users that already exist and to
    invite new ones by user_email, where the user_id is allocated server-side.
    """
    if user_api_key_dict.user_role in (
        LitellmUserRoles.PROXY_ADMIN,
        LitellmUserRoles.PROXY_ADMIN.value,
    ):
        return

    unknown_user_ids: Final = tuple(
        member.user_id for member in members if member.user_id is not None and member.user_id not in existing_user_ids
    )
    if not unknown_user_ids:
        return

    listed: Final = ", ".join(unknown_user_ids[:_MAX_REPORTED_UNKNOWN_USER_IDS])
    remaining: Final = len(unknown_user_ids) - _MAX_REPORTED_UNKNOWN_USER_IDS
    raise HTTPException(
        status_code=403,
        detail={  # mutable-ok: HTTPException detail must be a plain mapping to keep this route's {"error": ...} response shape
            "error": (
                "Only proxy admins can add a user_id that does not exist yet: {}{}. "
                "Add the member by user_email to invite a new user, or ask a proxy admin "
                "to create the user first.".format(listed, f" and {remaining} more" if remaining > 0 else "")
            )
        },
    )


def _members_audit_value(members: Sequence[Member]) -> str:
    """Serialize a team's member list for an audit-log value.

    The audit-log columns hold a JSON object, so the member list is nested
    under a key rather than serialized as a top-level array.
    """
    return safe_dumps(
        {  # mutable-ok: the audit-log JSON column rejects a top-level array, so this value must be an object
            "members_with_roles": tuple(member.model_dump() for member in members)
        }
    )


async def _create_team_member_add_audit_logs(
    team_id: str,
    updated_users: Sequence[LiteLLM_UserTable],
    existing_user_ids: frozenset[str],
    before_members: Sequence[Member],
    after_members: Sequence[Member],
    user_api_key_dict: UserAPIKeyAuth,
    litellm_proxy_admin_name: str,
) -> None:
    """Record the membership change, and any user row it created, in the audit log.

    The entries are written concurrently so a request adding many members does
    not pay for them one after another.
    """
    from litellm.proxy.management_helpers.audit_logs import create_object_audit_log

    created_user_entries: Final = tuple(
        create_object_audit_log(
            object_id=user.user_id,
            action="created",
            litellm_changed_by=None,
            user_api_key_dict=user_api_key_dict,
            litellm_proxy_admin_name=litellm_proxy_admin_name,
            table_name=LitellmTableNames.USER_TABLE_NAME,
            before_value=None,
            after_value=safe_dumps(user.model_dump(exclude_none=True)),
        )
        for user in updated_users
        if user.user_id is not None and user.user_id not in existing_user_ids
    )

    membership_entry: Final = create_object_audit_log(
        object_id=team_id,
        action="updated",
        litellm_changed_by=None,
        user_api_key_dict=user_api_key_dict,
        litellm_proxy_admin_name=litellm_proxy_admin_name,
        table_name=LitellmTableNames.TEAM_TABLE_NAME,
        before_value=_members_audit_value(before_members),
        after_value=_members_audit_value(after_members),
    )

    await asyncio.gather(*created_user_entries, membership_entry)


async def _validate_and_populate_member_user_info(
    member: Member,
    prisma_client: PrismaClient,
) -> Member:
    """
    Validate and populate user_email/user_id for a member.

    Logic:
    1. If both user_email and user_id are provided, verify they belong to the same user (use user_email as source of truth)
    2. If only user_email is provided, populate user_id from DB
    3. If only user_id is provided, populate user_email from DB (if user exists)
    4. If only user_id is provided and doesn't exist, allow it to pass with user_email as None (will be upserted later)
    5. If user_email and user_id mismatch, throw error

    Returns a Member with user_email and user_id populated (user_email may be None if only user_id provided and user doesn't exist).
    """
    if member.user_email is None and member.user_id is None:
        raise HTTPException(
            status_code=400,
            detail={"error": "Either user_id or user_email must be provided"},
        )

    # Case 1: Both user_email and user_id provided - verify they match
    if member.user_email is not None and member.user_id is not None:
        # Use user_email as source of truth
        # Check for multiple users with same email first
        users_by_email = await prisma_client.get_data(
            key_val={"user_email": member.user_email},
            table_name="user",
            query_type="find_all",
        )

        if users_by_email is None or (isinstance(users_by_email, list) and len(users_by_email) == 0):
            # User doesn't exist yet - this is fine, will be created later
            return member

        if isinstance(users_by_email, list) and len(users_by_email) > 1:
            raise HTTPException(
                status_code=400,
                detail={
                    "error": f"Multiple users found with email '{member.user_email}'. Please use 'user_id' instead."
                },
            )

        # Get the single user
        user_by_email = users_by_email[0]

        # Verify the user_id matches
        if user_by_email.user_id != member.user_id:
            raise HTTPException(
                status_code=400,
                detail={
                    "error": f"user_email '{member.user_email}' and user_id '{member.user_id}' do not belong to the same user."
                },
            )

        # Both match, return as is
        return member

    # Case 2: Only user_email provided - populate user_id from DB
    if member.user_email is not None and member.user_id is None:
        user_by_email = await _user_db(prisma_client).find_first(
            where={"user_email": {"equals": member.user_email, "mode": "insensitive"}}
        )

        if user_by_email is None:
            # User doesn't exist yet - this is fine, will be created later
            return member

        # Check for multiple users with same email
        users_by_email = await prisma_client.get_data(
            key_val={"user_email": member.user_email},
            table_name="user",
            query_type="find_all",
        )

        if users_by_email and isinstance(users_by_email, list) and len(users_by_email) > 1:
            raise HTTPException(
                status_code=400,
                detail={
                    "error": f"Multiple users found with email '{member.user_email}'. Please use 'user_id' instead."
                },
            )

        # Populate user_id
        member.user_id = user_by_email.user_id
        return member

    # Case 3: Only user_id provided - populate user_email from DB if user exists
    if member.user_id is not None and member.user_email is None:
        user_by_id: Final = await _user_db(prisma_client).find_unique(where={"user_id": member.user_id})

        if user_by_id is None:
            # User doesn't exist yet - allow it to pass with user_email as None
            # Will be upserted later with just user_id and null email
            return member

        # Populate user_email
        member.user_email = user_by_id.user_email
        return member

    return member


@router.post(
    "/team/member_add",
    tags=["team management"],
    dependencies=[Depends(user_api_key_auth)],
    response_model=TeamAddMemberResponse,
)
@management_endpoint_wrapper
async def team_member_add(
    data: TeamMemberAddRequest,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    Add new members (either via user_email or user_id) to a team

    If user doesn't exist, new user row will also be added to User Table

    Only proxy_admin or admin of team, allowed to access this endpoint.
    ```

    curl -X POST 'http://0.0.0.0:4000/team/member_add' \
    -H 'Authorization: Bearer sk-1234' \
    -H 'Content-Type: application/json' \
    -d '{"team_id": "45e3e396-ee08-4a61-a88e-16b3ce7e0849", "member": {"role": "user", "user_id": "krrish247652@berri.ai"}}'

    ```
    """
    from litellm.proxy.proxy_server import (
        litellm_proxy_admin_name,
        premium_user,
        prisma_client,
        proxy_logging_obj,
        user_api_key_cache,
    )

    try:
        team_call_validation_checks(
            prisma_client=prisma_client,
            data=data,
            premium_user=premium_user,
        )
    except HTTPException as e:
        raise e

    validate_budget_duration(data.budget_duration)

    prisma_client = cast(PrismaClient, prisma_client)

    existing_team_row: Final = await get_team_object(
        team_id=data.team_id,
        prisma_client=prisma_client,
        user_api_key_cache=user_api_key_cache,
        parent_otel_span=None,
        proxy_logging_obj=proxy_logging_obj,
        check_cache_only=False,
        check_db_only=True,
    )
    if existing_team_row is None:
        raise HTTPException(
            status_code=404,
            detail={"error": f"Team not found for team_id={getattr(data, 'team_id', None)}"},
        )

    complete_team_data: Final = LiteLLM_TeamTable.model_validate(existing_team_row.model_dump())

    team_member_add_duplication_check(
        data=data,
        existing_team_row=complete_team_data,
    )

    # Validate permissions
    await _validate_team_member_add_permissions(
        user_api_key_dict=user_api_key_dict,
        complete_team_data=complete_team_data,
        data=data,
    )

    requested_members: Final = tuple(data.member) if isinstance(data.member, list) else (data.member,)
    caller_supplied_user_ids = frozenset(member.user_id for member in requested_members if member.user_id is not None)
    existing_user_ids: Final = await _resolve_existing_member_user_ids(
        members=requested_members,
        prisma_client=prisma_client,
    )
    _validate_member_user_id_provisioning(
        members=requested_members,
        existing_user_ids=existing_user_ids,
        user_api_key_dict=user_api_key_dict,
    )
    members_before_add: Final = tuple(complete_team_data.members_with_roles)

    # Validate and populate user_email/user_id for members before processing
    if isinstance(data.member, Member):
        await _validate_and_populate_member_user_info(
            member=data.member,
            prisma_client=prisma_client,
        )
    elif isinstance(data.member, list):
        for m in data.member:
            await _validate_and_populate_member_user_info(
                member=m,
                prisma_client=prisma_client,
            )

    pre_existing_user_ids: Final = _pre_existing_user_ids(
        members=requested_members,
        caller_supplied_user_ids=caller_supplied_user_ids,
        existing_user_ids=existing_user_ids,
    )

    (
        updated_team,
        updated_users,
        updated_team_memberships,
    ) = await _add_team_members_to_team(
        data=data,
        complete_team_data=complete_team_data,
        prisma_client=prisma_client,
        user_api_key_dict=user_api_key_dict,
        litellm_proxy_admin_name=litellm_proxy_admin_name,
    )

    _emit_team_members_metric(complete_team_data)

    await _create_team_member_add_audit_logs(
        team_id=data.team_id,
        updated_users=updated_users,
        existing_user_ids=pre_existing_user_ids,
        before_members=members_before_add,
        after_members=tuple(complete_team_data.members_with_roles),
        user_api_key_dict=user_api_key_dict,
        litellm_proxy_admin_name=litellm_proxy_admin_name,
    )

    return TeamAddMemberResponse.model_validate(
        {
            **updated_team.model_dump(),
            "updated_users": updated_users,
            "updated_team_memberships": updated_team_memberships,
        }
    )


def _is_member_addressed_by(member: Member, data: TeamMemberDeleteRequest) -> bool:
    return (data.user_id is not None and member.user_id is not None and data.user_id == member.user_id) or (
        data.user_email is not None and member.user_email is not None and data.user_email == member.user_email
    )


def _cleanup_members_with_roles(
    existing_team_row: LiteLLM_TeamTable,
    data: TeamMemberDeleteRequest,
) -> tuple[tuple[Member, ...], list[Member]]:
    """Split a team's members_with_roles into the entries the request addresses and the ones that stay.

    The addressed entries are returned rather than a bare found/not-found flag because they carry the
    user_id the request may not have supplied, and every cleanup that keys off the user rather than
    off the roster has to run against that id.
    """
    removed_team_members: Final = tuple(
        m for m in existing_team_row.members_with_roles if _is_member_addressed_by(m, data)
    )
    new_team_members: Final = [m for m in existing_team_row.members_with_roles if not _is_member_addressed_by(m, data)]
    return removed_team_members, new_team_members


@router.post(
    "/team/member_delete",
    tags=["team management"],
    dependencies=[Depends(user_api_key_auth)],
)
@management_endpoint_wrapper
async def team_member_delete(
    data: TeamMemberDeleteRequest,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    [BETA]

    delete members (either via user_email or user_id) from a team

    If user doesn't exist, an exception will be raised
    ```
    curl -X POST 'http://0.0.0.0:8000/team/member_delete' \

    -H 'Authorization: Bearer sk-1234' \

    -H 'Content-Type: application/json' \

    -d '{
        "team_id": "45e3e396-ee08-4a61-a88e-16b3ce7e0849",
        "user_id": "krrish247652@berri.ai"
    }'
    ```
    """
    from litellm.proxy.proxy_server import prisma_client

    if prisma_client is None:
        raise HTTPException(status_code=500, detail={"error": "No db connected"})

    if data.team_id is None:
        raise HTTPException(status_code=400, detail={"error": "No team id passed in"})

    if data.user_id is None and data.user_email is None:
        raise HTTPException(
            status_code=400,
            detail={"error": "Either user_id or user_email needs to be passed in"},
        )

    _existing_team_row: Final = await _team_db(prisma_client).find_unique(where={"team_id": data.team_id})

    if _existing_team_row is None:
        raise HTTPException(
            status_code=400,
            detail={"error": f"Team id={data.team_id} does not exist in db"},
        )
    existing_team_row: Final = LiteLLM_TeamTable.model_validate(_existing_team_row.model_dump())

    ## CHECK IF USER IS PROXY ADMIN OR TEAM ADMIN OR ORG ADMIN

    if (
        user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value
        and not _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=existing_team_row)
        and not await _is_user_org_admin_for_team(user_api_key_dict=user_api_key_dict, team_obj=existing_team_row)
    ):
        raise HTTPException(
            status_code=403,
            detail={
                "error": "Call not allowed. User not proxy admin OR team admin. route={}, team_id={}".format(
                    "/team/member_delete", existing_team_row.team_id
                )
            },
        )

    ## DELETE MEMBER FROM TEAM
    # Everything from here on runs under the team's advisory lock, the same one
    # /team/member_add and /team/delete take: without it, this endpoint's own row-level
    # update lock used to be the only thing serializing it against a concurrent member_add,
    # and only by accident (their SELECT ... FOR UPDATE contended for the same row lock this
    # UPDATE takes). Now that member_add reads under the advisory lock instead, this has to
    # take it too, and re-read the roster under it rather than off the snapshot validated
    # above, or a member_add that commits in between can have its addition silently
    # overwritten by this delete computing from stale data.
    async with prisma_client.tx() as tx:
        await tx.query_raw(TEAM_ADVISORY_LOCK_SQL, data.team_id)

        fresh_members: Final = await TeamRepository(prisma_client).get_members_with_roles_locked(tx, data.team_id)
        if fresh_members is None:
            raise HTTPException(
                status_code=400,
                detail={"error": f"Team id={data.team_id} does not exist in db"},
            )

        removed_team_members, new_team_members = _cleanup_members_with_roles(
            existing_team_row=LiteLLM_TeamTable(team_id=data.team_id, members_with_roles=fresh_members),
            data=data,
        )

        if not removed_team_members:
            raise HTTPException(status_code=400, detail={"error": "User not found in team"})

        existing_team_row.members_with_roles = new_team_members

        _db_new_team_members: Final[list[dict]] = [m.model_dump() for m in new_team_members]

        ## DELETE TEAM ID from USER ROW, IF EXISTS ##
        # get user row
        removed_user_ids: Final = frozenset(m.user_id for m in removed_team_members if m.user_id is not None)
        key_val: Final[Mapping[str, object]] = (
            {"user_id": {"in": sorted(removed_user_ids)}} if removed_user_ids else {"user_email": data.user_email}
        )
        member_tx: Final[_MemberDeleteTx] = tx
        existing_user_rows: Final = await member_tx.litellm_usertable.find_many(where=key_val)

        # Also clean up any existing team membership rows for this user and team
        user_ids_to_delete: Final = removed_user_ids.union(
            (data.user_id,) if data.user_id is not None else (),
            (user.user_id for user in existing_user_rows if user.user_id),
        )

        ## DELETE KEYS CREATED BY USER FOR THIS TEAM
        # Fetch keys before deletion so their audit records can be persisted alongside the delete.
        # An empty user_ids_to_delete still resolves cleanly: prisma's "in": [] matches no rows.
        keys_to_delete: Final = await member_tx.litellm_verificationtoken.find_many(
            where={
                "user_id": {"in": sorted(user_ids_to_delete)},
                "team_id": data.team_id,
            }
        )

        await _team_tx_db(tx).update(
            where={"team_id": data.team_id},
            data={"members_with_roles": json.dumps(_db_new_team_members)},
        )

        for existing_user in existing_user_rows:
            if data.team_id in existing_user.teams:
                await tx.litellm_usertable.update(
                    where={"user_id": existing_user.user_id},
                    data={"teams": {"set": [team for team in existing_user.teams if team != data.team_id]}},
                )

        for _uid in sorted(user_ids_to_delete):
            await tx.litellm_teammembership.delete_many(where={"team_id": data.team_id, "user_id": _uid})

        if user_ids_to_delete:
            if keys_to_delete:
                from litellm.proxy.management_endpoints.key_management_endpoints import (
                    _persist_deleted_verification_tokens,
                )

                await _persist_deleted_verification_tokens(
                    keys=keys_to_delete,
                    prisma_client=prisma_client,
                    user_api_key_dict=user_api_key_dict,
                    litellm_changed_by=None,
                    tx=tx,
                )

            await tx.litellm_verificationtoken.delete_many(
                where={
                    "user_id": {"in": sorted(user_ids_to_delete)},
                    "team_id": data.team_id,
                }
            )

    _emit_team_members_metric(existing_team_row)

    return existing_team_row


_MEMBER_BUDGET_PATCH_FIELDS: Final = {
    "max_budget_in_team": "max_budget",
    "tpm_limit": "tpm_limit",
    "rpm_limit": "rpm_limit",
    "budget_duration": "budget_duration",
    "allowed_models": "allowed_models",
}


def _build_member_budget_patch(data: TeamMemberUpdateRequest) -> dict[str, object]:
    """Map the budget fields the request actually set (merge-patch: a sent
    value updates, an explicit null clears, an absent field is left untouched)
    to their budget-table columns."""
    provided: Final = data.model_dump(exclude_unset=True)
    return {
        column: provided[request_field]
        for request_field, column in _MEMBER_BUDGET_PATCH_FIELDS.items()
        if request_field in provided
    }


@router.post(
    "/team/member_update",
    tags=["team management"],
    dependencies=[Depends(user_api_key_auth)],
    response_model=TeamMemberUpdateResponse,
)
@management_endpoint_wrapper
async def team_member_update(
    data: TeamMemberUpdateRequest,
    http_request: Request,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    [BETA]

    Update team member budgets and team member role
    """
    from litellm.proxy.proxy_server import premium_user, prisma_client, user_api_key_cache

    if prisma_client is None:
        raise HTTPException(status_code=500, detail={"error": "No db connected"})

    if data.team_id is None:
        raise HTTPException(status_code=400, detail={"error": "No team id passed in"})

    if data.role == "admin" and not premium_user:
        # exactly the same text your proxy throws for add:
        raise HTTPException(
            status_code=400,
            detail="Assigning team admins is a premium feature. You must be a LiteLLM Enterprise user to use this feature. If you have a license please set `LITELLM_LICENSE` in your env. Get a 7 day trial key here: https://www.litellm.ai/#trial. Pricing: https://www.litellm.ai/#pricing",
        )
    if data.user_id is None and data.user_email is None:
        raise HTTPException(
            status_code=400,
            detail={"error": "Either user_id or user_email needs to be passed in"},
        )

    validate_budget_duration(data.budget_duration)

    _existing_team_row: Final = await _team_db(prisma_client).find_unique(where={"team_id": data.team_id})

    if _existing_team_row is None:
        raise HTTPException(
            status_code=400,
            detail={"error": f"Team id={data.team_id} does not exist in db"},
        )
    existing_team_row: Final = LiteLLM_TeamTable.model_validate(_existing_team_row.model_dump())

    ## CHECK IF USER IS PROXY ADMIN OR TEAM ADMIN OR ORG ADMIN

    if (
        user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value
        and not _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=existing_team_row)
        and not await _is_user_org_admin_for_team(user_api_key_dict=user_api_key_dict, team_obj=existing_team_row)
    ):
        raise HTTPException(
            status_code=403,
            detail={
                "error": "Call not allowed. User not proxy admin OR team admin. route={}, team_id={}".format(
                    "/team/member_delete", existing_team_row.team_id
                )
            },
        )

    returned_team_info: Final[TeamInfoResponseObject] = await team_info(
        http_request=http_request,
        team_id=data.team_id,
        key_limit=None,
        user_api_key_dict=user_api_key_dict,
    )

    team_table: Final = returned_team_info["team_info"]

    ## get user id
    received_user_id: str | None = None
    if data.user_id is not None:
        received_user_id = data.user_id
    elif data.user_email is not None:
        for member in returned_team_info["team_info"].members_with_roles:
            if member.user_email is not None and member.user_email == data.user_email:
                received_user_id = member.user_id
                break

    if received_user_id is None:
        raise HTTPException(
            status_code=400,
            detail={"error": f"User id doesn't exist in team table. Data={data}"},
        )
    ## find the relevant team membership
    identified_budget_id: str | None = None
    for tm in returned_team_info["team_memberships"]:
        if tm.user_id == received_user_id:
            identified_budget_id = tm.budget_id
            break

    # If this membership still points at the team's shared default member
    # budget, _upsert_budget_and_membership will clone-on-write so that the
    # update only touches this user (not every member sharing the default).
    team_default_budget_id: str | None = None
    if team_table.metadata is not None:
        raw_default_budget_id: Final = team_table.metadata.get("team_member_budget_id")
        if isinstance(raw_default_budget_id, str):
            team_default_budget_id = raw_default_budget_id

    ### upsert new budget
    budget_patch: Final = _build_member_budget_patch(data)
    async with prisma_client.tx() as tx:
        await _upsert_budget_and_membership(
            tx=tx,
            team_id=data.team_id,
            user_id=received_user_id,
            existing_budget_id=identified_budget_id,
            user_api_key_dict=user_api_key_dict,
            budget_patch=budget_patch,
            team_default_budget_id=team_default_budget_id,
        )
    if budget_patch:
        await invalidate_team_member_spend_state(
            user_id=received_user_id,
            team_id=data.team_id,
            user_api_key_cache=user_api_key_cache,
        )

    ### update team member role
    if data.role is not None:
        team_members: Final[list[Member]] = []
        for member in team_table.members_with_roles:
            if member.user_id == received_user_id:
                team_members.append(
                    Member(
                        user_id=member.user_id,
                        role=data.role,
                        user_email=data.user_email or member.user_email,
                    )
                )
            else:
                team_members.append(member)

        team_table.members_with_roles = team_members

        _db_team_members: Final[list[dict]] = [m.model_dump() for m in team_members]
        await _team_db(prisma_client).update(
            where={"team_id": data.team_id},
            data={"members_with_roles": json.dumps(_db_team_members)},
        )

    return TeamMemberUpdateResponse(
        team_id=data.team_id,
        user_id=received_user_id,
        user_email=data.user_email,
        max_budget_in_team=data.max_budget_in_team,
        tpm_limit=data.tpm_limit,
        rpm_limit=data.rpm_limit,
        budget_duration=data.budget_duration,
        allowed_models=data.allowed_models,
    )


def _check_not_resetting_own_spend(user_id: str, user_api_key_dict: UserAPIKeyAuth) -> None:
    """
    _verify_team_access authorizes a team admin (or org admin) over their own
    team, with no check that the target user_id differs from the caller. Left
    unchecked, that admin could target their own LiteLLM_TeamMembership row and
    repeatedly reset it to 0 right before it crosses their per-member cap,
    consuming the shared team budget without the configured limit ever binding.
    Only a proxy admin may reset an admin's own spend.
    """
    if user_id == user_api_key_dict.user_id and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN:
        _raise_reset_spend_error(status.HTTP_403_FORBIDDEN, "Cannot reset your own spend. Ask a proxy admin.")


def _raise_reset_spend_error(status_code: int, message: str) -> NoReturn:
    detail: Final = {"error": message}  # mutable-ok: HTTPException.detail takes a dict
    raise HTTPException(status_code=status_code, detail=detail)


def _validate_team_member_reset_spend_value(
    reset_to: object,
    membership: LiteLLM_TeamMembership,
) -> float:
    if not isinstance(reset_to, (int, float)):
        _raise_reset_spend_error(status.HTTP_400_BAD_REQUEST, "reset_to must be a float")

    reset_to_float: Final = float(reset_to)
    if not math.isfinite(reset_to_float) or reset_to_float < 0:
        _raise_reset_spend_error(status.HTTP_400_BAD_REQUEST, "reset_to must be a finite number >= 0")

    current_spend: Final = membership.spend or 0.0
    if reset_to_float > current_spend:
        _raise_reset_spend_error(
            status.HTTP_400_BAD_REQUEST,
            f"reset_to ({reset_to_float}) must be <= current spend ({current_spend})",
        )

    max_budget: Final = membership.litellm_budget_table.max_budget if membership.litellm_budget_table else None
    if max_budget is not None and reset_to_float > max_budget:
        _raise_reset_spend_error(
            status.HTTP_400_BAD_REQUEST,
            f"reset_to ({reset_to_float}) must be <= budget ({max_budget})",
        )

    return reset_to_float


@router.post(
    "/team/{team_id}/member/{user_id}/reset_spend",
    tags=["team management"],  # mutable-ok: FastAPI's `tags` param is typed as list[str], not Sequence
    dependencies=(Depends(user_api_key_auth),),
)
@management_endpoint_wrapper
async def reset_team_member_spend_fn(
    team_id: str,
    user_id: str,
    data: ResetSpendRequest,
    user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
):
    """
    Reset a team member's tracked spend against their per-member budget.

    A member's spend is tracked separately from both their own personal
    budget and the team's own budget (LiteLLM_TeamMembership.spend), so
    neither /user/update nor /team/update can clear it: this is the only
    endpoint that does. The cross-pod spend counter and cached membership
    reads are invalidated so the reset takes effect on the member's next
    request rather than waiting on the membership cache's TTL.
    """
    from litellm.proxy.proxy_server import prisma_client, proxy_logging_obj, user_api_key_cache

    if prisma_client is None:
        _raise_reset_spend_error(status.HTTP_500_INTERNAL_SERVER_ERROR, "DB not connected. prisma_client is None")

    team_obj: Final = await get_team_object(
        team_id=team_id,
        prisma_client=prisma_client,
        user_api_key_cache=user_api_key_cache,
        parent_otel_span=None,
        proxy_logging_obj=proxy_logging_obj,
        check_db_only=True,
    )
    await _verify_team_access(team_obj=team_obj, user_api_key_dict=user_api_key_dict)
    _check_not_resetting_own_spend(user_id=user_id, user_api_key_dict=user_api_key_dict)

    membership_where: Final = {  # mutable-ok: prisma client requires a plain dict where= argument
        "user_id_team_id": {"user_id": user_id, "team_id": team_id}  # mutable-ok: same prisma where= argument
    }
    _membership_row: Final = await _team_membership_db(prisma_client).find_unique(
        where=membership_where,
        include={"litellm_budget_table": True},  # mutable-ok: prisma client requires a plain dict include= argument
    )
    if _membership_row is None:
        _raise_reset_spend_error(status.HTTP_404_NOT_FOUND, f"User {user_id} is not a member of team {team_id}.")
    membership: Final = LiteLLM_TeamMembership.model_validate(_membership_row.model_dump())

    current_spend: Final = membership.spend or 0.0
    reset_to: Final = _validate_team_member_reset_spend_value(data.reset_to, membership)

    await _team_membership_db(prisma_client).update(
        where=membership_where,
        data={"spend": reset_to},  # mutable-ok: prisma client requires a plain dict data= argument
    )

    await invalidate_team_member_spend_state(
        user_id=user_id,
        team_id=team_id,
        user_api_key_cache=user_api_key_cache,
        new_spend=reset_to,
    )

    return {  # mutable-ok: matches this router's established untyped-response-dict convention
        "team_id": team_id,
        "user_id": user_id,
        "spend": reset_to,
        "previous_spend": current_spend,
        "max_budget": membership.litellm_budget_table.max_budget if membership.litellm_budget_table else None,
    }


def _create_results_from_response(
    members: list[Member],
    response: TeamAddMemberResponse,
) -> list[TeamMemberAddResult]:
    """
    Convert TeamAddMemberResponse into individual TeamMemberAddResult objects
    """
    results: Final[list[TeamMemberAddResult]] = []

    for member in members:
        # Find corresponding updated user
        updated_user = None
        for user in response.updated_users:
            if (member.user_id and user.user_id == member.user_id) or (
                member.user_email and user.user_email == member.user_email
            ):
                updated_user = user.model_dump()
                break

        # Find corresponding updated team membership
        updated_team_membership = None
        for tm in response.updated_team_memberships:
            if member.user_id and tm.user_id == member.user_id:
                updated_team_membership = tm.model_dump()
                break

        results.append(
            TeamMemberAddResult(
                user_id=member.user_id,
                user_email=member.user_email,
                success=True,
                updated_user=updated_user,
                updated_team_membership=updated_team_membership,
            )
        )

    return results


@router.post(
    "/team/bulk_member_add",
    tags=["team management"],
    dependencies=[Depends(user_api_key_auth)],
    response_model=BulkTeamMemberAddResponse,
)
@management_endpoint_wrapper
async def bulk_team_member_add(
    data: BulkTeamMemberAddRequest,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    Bulk add multiple members to a team at once.
    
    This endpoint reuses the same logic as /team/member_add but provides a bulk-friendly response format.
    
    Parameters:
    - team_id: str - The ID of the team to add members to
    - members: List[Member] - List of members to add to the team
    - all_users: Optional[bool] - Flag to add all users on Proxy to the team
    - max_budget_in_team: Optional[float] - Maximum budget allocated to each user within the team
    
    Returns:
    - results: List of individual member addition results
    - total_requested: Total number of members requested for addition
    - successful_additions: Number of successful additions  
    - failed_additions: Number of failed additions
    - updated_team: The updated team object
    
    Example request:
    ```bash
    curl --location 'http://0.0.0.0:4000/team/bulk_member_add' \
    --header 'Authorization: Bearer sk-1234' \
    --header 'Content-Type: application/json' \
    --data '{
        "team_id": "team-1234",
        "members": [
            {
                "user_id": "user1",
                "role": "user"
            },
            {
                "user_email": "user2@example.com",
                "role": "admin"
            }
        ],
        "max_budget_in_team": 100.0
    }'
    ```
    """
    from litellm.proxy._types import CommonProxyErrors
    from litellm.proxy.proxy_server import prisma_client

    if prisma_client is None:
        raise HTTPException(
            status_code=500,
            detail={"error": CommonProxyErrors.db_not_connected_error.value},
        )

    if data.all_users:
        # `all_users=True` pulls every user in the database into this team,
        # regardless of org. Any team admin could use it to capture every
        # user across every org into a team they control. Restrict to
        # PROXY_ADMIN.
        if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value:
            raise HTTPException(
                status_code=403,
                detail={
                    "error": (
                        "`all_users=true` is restricted to PROXY_ADMIN. "
                        "Org/team admins must specify explicit member lists."
                    )
                },
            )
        # get all users from the database
        all_users_in_db: Final = await _user_db(prisma_client).find_many(order={"created_at": "desc"})
        data.members = [
            Member(
                user_id=user.user_id,
                user_email=user.user_email,
                role="user",
            )
            for user in all_users_in_db
        ]

    if not data.members:
        raise HTTPException(
            status_code=400,
            detail={"error": "At least one member is required"},
        )

    # Limit batch size to prevent overwhelming the system
    MAX_BATCH_SIZE: Final = 500
    if len(data.members) > MAX_BATCH_SIZE:
        raise HTTPException(
            status_code=400,
            detail={"error": f"Maximum {MAX_BATCH_SIZE} members can be added at once"},
        )

    try:
        # Reuse the existing team_member_add logic directly
        response: Final = await team_member_add(
            data=TeamMemberAddRequest(
                team_id=data.team_id,
                member=data.members,  # Pass the entire list
                max_budget_in_team=data.max_budget_in_team,
            ),
            user_api_key_dict=user_api_key_dict,
        )

        # Convert to bulk response format
        results = _create_results_from_response(data.members, response)

        return BulkTeamMemberAddResponse(
            team_id=data.team_id,
            results=results,
            total_requested=len(data.members),
            successful_additions=len(results),  # All succeeded if we got here
            failed_additions=0,
            updated_team=response.model_dump(),
        )

    except Exception as e:
        # If the entire operation fails, mark all members as failed
        verbose_proxy_logger.exception(e)
        error_message: Final = str(e)
        results = [
            TeamMemberAddResult(
                user_id=member.user_id,
                user_email=member.user_email,
                success=False,
                error=error_message,
            )
            for member in data.members
        ]

        return BulkTeamMemberAddResponse(
            team_id=data.team_id,
            results=results,
            total_requested=len(data.members),
            successful_additions=0,
            failed_additions=len(data.members),
            updated_team=None,
        )


@router.post("/team/delete", tags=["team management"], dependencies=[Depends(user_api_key_auth)])
@management_endpoint_wrapper
async def delete_team(
    data: DeleteTeamRequest,
    http_request: Request,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
    litellm_changed_by: str | None = Header(
        None,
        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",
    ),
):
    """
    delete team and associated team keys

    Parameters:
    - team_ids: List[str] - Required. List of team IDs to delete. Example: ["team-1234", "team-5678"]

    ```
    curl --location 'http://0.0.0.0:4000/team/delete' \
    --header 'Authorization: Bearer sk-1234' \
    --header 'Content-Type: application/json' \
    --data-raw '{
        "team_ids": ["8d916b1c-510d-4894-a334-1c16a93344f5"]
    }'
    ```
    """
    from litellm.proxy.management_helpers.audit_logs import (
        get_audit_log_changed_by,
        is_audit_logging_enabled,
    )
    from litellm.proxy.proxy_server import (
        create_audit_log_for_update,
        litellm_proxy_admin_name,
        prisma_client,
        proxy_logging_obj,
        user_api_key_cache,
    )

    if prisma_client is None:
        raise HTTPException(status_code=500, detail={"error": "No db connected"})

    if data.team_ids is None:
        raise HTTPException(status_code=400, detail={"error": "No team id passed in"})

    # check that all teams passed exist
    team_rows: Final[list[LiteLLM_TeamTable]] = []
    for team_id in data.team_ids:
        try:
            team_row_base: BaseModel | None = await _team_db(prisma_client).find_unique(where={"team_id": team_id})
            if team_row_base is None:
                raise Exception
        except Exception:
            raise HTTPException(
                status_code=404,
                detail={"error": f"Team not found, passed team_id={team_id}"},
            )
        team_row_pydantic = LiteLLM_TeamTable.model_validate(team_row_base.model_dump())

        # Verify caller has access to manage this team
        await _verify_team_access(
            team_obj=team_row_pydantic,
            user_api_key_dict=user_api_key_dict,
        )

        team_rows.append(team_row_pydantic)

    await _persist_deleted_team_records(
        teams=team_rows,
        prisma_client=prisma_client,
        user_api_key_dict=user_api_key_dict,
        litellm_changed_by=litellm_changed_by,
    )

    # we do this after the first for loop, since first for loop is for validation. we only want this inserted after validation passes
    if is_audit_logging_enabled():
        # make an audit log for each team deleted
        for team_id in data.team_ids:
            team_row: LiteLLM_TeamTable | None = await prisma_client.get_data(
                team_id=team_id, table_name="team", query_type="find_unique"
            )

            if team_row is None:
                continue

            _team_row = team_row.json(exclude_none=True)

            asyncio.create_task(
                create_audit_log_for_update(
                    request_data=LiteLLM_AuditLogs(
                        id=str(uuid.uuid4()),
                        updated_at=datetime.now(timezone.utc),
                        changed_by=get_audit_log_changed_by(
                            litellm_changed_by=litellm_changed_by,
                            user_api_key_dict=user_api_key_dict,
                            litellm_proxy_admin_name=litellm_proxy_admin_name,
                        ),
                        changed_by_api_key=user_api_key_dict.api_key,
                        table_name=LitellmTableNames.TEAM_TABLE_NAME,
                        object_id=team_id,
                        action="deleted",
                        updated_values="{}",
                        before_value=_team_row,
                    )
                )
            )

    # End of Audit logging

    ## DELETE ASSOCIATED KEYS
    # Fetch keys before deletion to persist them
    from litellm.proxy.management_endpoints.key_management_endpoints import (
        _persist_deleted_verification_tokens,
    )

    keys_to_delete: Final = await _tokens_db(prisma_client).find_many(where={"team_id": {"in": data.team_ids}})

    if keys_to_delete:
        await _persist_deleted_verification_tokens(
            keys=keys_to_delete,
            prisma_client=prisma_client,
            user_api_key_dict=user_api_key_dict,
            litellm_changed_by=litellm_changed_by,
        )

    await prisma_client.delete_data(team_id_list=data.team_ids, table_name="key")

    await _invalidate_deleted_key_cache(
        keys=keys_to_delete,
        user_api_key_cache=user_api_key_cache,
        proxy_logging_obj=proxy_logging_obj,
    )

    ## DELETE ASSOCIATED BYOK MODELS
    # Runs before the team rows are deleted so a mid-flight failure never leaves
    # the team gone with its models orphaned.
    from litellm.proxy.management_endpoints.model_management_endpoints import (
        delete_team_models,
    )
    from litellm.proxy.proxy_server import llm_router

    await delete_team_models(
        team_ids=data.team_ids,
        prisma_client=prisma_client,
        llm_router=llm_router,
    )

    # ## DELETE TEAM MEMBERSHIPS
    for team_row in team_rows:
        ### get all team members
        team_members = team_row.members_with_roles
        ### call team_member_delete for each team member
        tasks = []
        for team_member in team_members:
            tasks.append(
                team_member_delete(
                    data=TeamMemberDeleteRequest(
                        team_id=team_row.team_id,
                        user_id=team_member.user_id,
                        user_email=team_member.user_email,
                    ),
                    user_api_key_dict=user_api_key_dict,
                )
            )
        await asyncio.gather(*tasks)

    await _sweep_deleted_team_references(team_ids=data.team_ids, prisma_client=prisma_client)

    ## DELETE TEAMS
    # Both the delete and the reconcile sweep run under every team's advisory lock
    # (TEAM_ADVISORY_LOCK_SQL, the same one /team/member_add takes before its own writes),
    # sorted so two overlapping batch deletes always request their locks in the same order.
    # A member_add mid-flight for one of these teams either finishes its write and releases
    # the lock before this transaction starts, in which case this sweep reaches what it wrote,
    # or is still waiting on the lock, in which case its own re-read happens after this commits
    # and sees the row gone before it writes anything.
    delete_filter: Final[_TeamIdInFilter] = {"team_id": {"in": data.team_ids}}
    async with prisma_client.tx() as tx:
        for team_id in sorted(data.team_ids):
            await tx.query_raw(TEAM_ADVISORY_LOCK_SQL, team_id)
        await tx.litellm_teamtable.delete_many(where=delete_filter)
        await _sweep_deleted_team_references_tx(team_ids=data.team_ids, tx=tx)

    deleted_teams: Final[_DeletedTeamsResult] = {"deleted_teams": data.team_ids}

    # Evict AFTER the rows are gone. Both writers of these keys (`_cache_team_object` and
    # `get_team_object_by_alias`) hydrate from the db, so evicting first leaves a window where a
    # concurrent auth lookup re-caches the still-present team and the delete looks like it never
    # invalidated anything. Nothing fallible runs between the delete and this, or a failure there
    # would strand the deleted team in cache.
    await _invalidate_deleted_team_cache(
        teams=team_rows,
        user_api_key_cache=user_api_key_cache,
        proxy_logging_obj=proxy_logging_obj,
    )

    for deleted_team in team_rows:
        await sync_team_access_group_membership(prisma_client=prisma_client, team_id=deleted_team.team_id)

    return deleted_teams


async def _sweep_deleted_team_references(team_ids: Sequence[str], prisma_client: PrismaClient) -> None:
    """
    Strip the deleted team ids from every user row and team-membership row that still references them.

    The per-member `team_member_delete` pass above only reaches users listed in the team's
    `members_with_roles`, so a user row that outlived its roster entry is invisible to it and keeps
    surfacing the team on `/user/info` after the team is gone.

    #36839 closed the route that created that drift, by resolving member removal off the roster
    entry's `user_id` rather than the identifier the caller happened to pass. It does not backfill
    rows that already drifted, which is the state this was reported against, so the sweep still has
    to run on delete.

    `array_remove` rather than read-filter-write: rewriting the whole array from a snapshot read
    outside a transaction drops any team a concurrent `/team/member_add` appended in between.
    """
    for team_id in team_ids:
        _ = await prisma_client.db.execute_raw(_STRIP_DELETED_TEAM_FROM_USERS_SQL, team_id)

    _ = await _team_membership_db(prisma_client).delete_many(where=_TeamIdInFilter(team_id={"in": tuple(team_ids)}))


async def _sweep_deleted_team_references_tx(team_ids: Sequence[str], tx: _TeamDeleteTx) -> None:
    """Same sweep as `_sweep_deleted_team_references`, run on the transaction that holds
    every id's advisory lock and deletes the team rows, so it commits or rolls back with them."""
    for team_id in team_ids:
        _ = await tx.execute_raw(_STRIP_DELETED_TEAM_FROM_USERS_SQL, team_id)

    membership_filter: Final[_TeamIdInFilter] = {"team_id": {"in": tuple(team_ids)}}
    _ = await tx.litellm_teammembership.delete_many(where=membership_filter)


async def _invalidate_deleted_key_cache(
    keys: "Sequence[prisma_models.LiteLLM_VerificationToken]",
    user_api_key_cache: UserApiKeyCache,
    proxy_logging_obj: ProxyLogging,
) -> None:
    """
    Evict the auth cache entry for every key deleted along with the team.

    `/key/delete` evicts as it goes, but the bulk delete above writes straight to the db. Auth
    resolves a cached key object without re-reading the team, so a key belonging to a deleted team
    keeps buying access until its TTL expires.
    """
    await delete_cache_key_objects(
        hashed_tokens=tuple(key.token for key in keys),
        user_api_key_cache=user_api_key_cache,
        proxy_logging_obj=proxy_logging_obj,
    )


async def _invalidate_deleted_team_cache(
    teams: Sequence[LiteLLM_TeamTable],
    user_api_key_cache: UserApiKeyCache,
    proxy_logging_obj: ProxyLogging,
) -> None:
    _ = await asyncio.gather(
        *(
            delete_cache_team_object(
                team_id=team.team_id,
                team_alias=team.team_alias,
                user_api_key_cache=user_api_key_cache,
                proxy_logging_obj=proxy_logging_obj,
            )
            for team in teams
        )
    )


def _transform_teams_to_deleted_records(
    teams: list[LiteLLM_TeamTable],
    user_api_key_dict: UserAPIKeyAuth,
    litellm_changed_by: str | None = None,
) -> list[dict[str, object]]:
    """Transform teams into deleted team records ready for persistence."""
    if not teams:
        return []

    deleted_at: Final = datetime.now(timezone.utc)
    records: Final = []
    for team in teams:
        team_payload = team.model_dump()
        deleted_record = LiteLLM_DeletedTeamTable.model_validate(
            {
                **team_payload,
                "deleted_at": deleted_at,
                "deleted_by": user_api_key_dict.user_id,
                "deleted_by_api_key": user_api_key_dict.api_key,
                "litellm_changed_by": litellm_changed_by,
            }
        )
        record = deleted_record.model_dump()

        for json_field in [
            "members_with_roles",
            "metadata",
            "model_spend",
            "model_max_budget",
            "router_settings",
        ]:
            if json_field in record and record[json_field] is not None:
                record[json_field] = json.dumps(record[json_field])

        for rel_key in (
            "litellm_model_table",
            "object_permission",
            "id",
            "budget_limits",  # not in LiteLLM_DeletedTeamTable schema
            "default_team_member_models",  # not in LiteLLM_DeletedTeamTable schema
        ):
            record.pop(rel_key, None)

        records.append(record)

    return records


async def _save_deleted_team_records(
    records: list[dict[str, object]],
    prisma_client: PrismaClient,
) -> None:
    """Save deleted team records to the database."""
    if not records:
        return
    await _deleted_team_db(prisma_client).create_many(data=records)


async def _persist_deleted_team_records(
    teams: list[LiteLLM_TeamTable],
    prisma_client: PrismaClient,
    user_api_key_dict: UserAPIKeyAuth,
    litellm_changed_by: str | None = None,
) -> None:
    """Persist deleted team records by transforming and saving them."""
    records: Final = _transform_teams_to_deleted_records(
        teams=teams,
        user_api_key_dict=user_api_key_dict,
        litellm_changed_by=litellm_changed_by,
    )
    await _save_deleted_team_records(
        records=records,
        prisma_client=prisma_client,
    )


async def validate_membership(user_api_key_dict: UserAPIKeyAuth, team_table: LiteLLM_TeamTable):
    if (
        user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN.value
        or user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY.value
    ):
        return

    if user_api_key_dict.team_id == team_table.team_id:  # allow team keys to check their info
        return

    # Handle case where user_id is None (e.g., team key accessing different team)
    if user_api_key_dict.user_id is None:
        if user_api_key_dict.team_id is not None:
            raise HTTPException(
                status_code=403,
                detail={
                    "error": f"Team key for team={user_api_key_dict.team_id} not authorized to access this team={team_table.team_id}"
                },
            )
        else:
            raise HTTPException(
                status_code=403,
                detail={
                    "error": f"API key not authorized to access this team={team_table.team_id}. No user_id or team_id associated with this key."
                },
            )

    # Check direct team membership
    if user_api_key_dict.user_id in [m.user_id for m in team_table.members_with_roles]:
        return

    # Check if user is an org admin for the team's organization
    if await _is_user_org_admin_for_team(user_api_key_dict=user_api_key_dict, team_obj=team_table):
        return

    raise HTTPException(
        status_code=403,
        detail={"error": f"User={user_api_key_dict.user_id} not authorized to access this team={team_table.team_id}"},
    )


async def _add_team_member_budget_table(
    team_member_budget_id: str,
    prisma_client: PrismaClient,
    team_info_response_object: TeamInfoResponseObjectTeamTable,
) -> TeamInfoResponseObjectTeamTable:
    try:
        team_budget: Final = await _budget_db(prisma_client).find_unique(where={"budget_id": team_member_budget_id})
        return team_info_response_object.model_copy(update={"team_member_budget_table": team_budget})
    except Exception:
        verbose_proxy_logger.info(
            "Team member budget table not found, passed team_member_budget_id=%s", team_member_budget_id
        )

    return team_info_response_object


async def _hydrate_member_emails(
    prisma_client: PrismaClient,
    members: Sequence[Member],
) -> tuple[Member, ...]:
    """Fill in ``user_email`` for roster entries that were stored without one.

    ``members_with_roles`` is a denormalized snapshot written at add-time, so an entry
    stored with ``user_email=None`` keeps that null even once the user row has an email.
    Look the missing ones up in ``LiteLLM_UserTable`` (one indexed query) and fill them
    in. A stored email is never overwritten - the snapshot stays the source of truth
    wherever it has a value.
    """
    missing_user_ids: Final = frozenset(m.user_id for m in members if not m.user_email and m.user_id is not None)
    if not missing_user_ids:
        return tuple(members)

    user_rows: Final[Sequence[prisma_models.LiteLLM_UserTable]] = await _user_db(prisma_client).find_many(
        where={  # mutable-ok: Prisma query filters are dict-shaped
            "user_id": {  # mutable-ok: Prisma query filters are dict-shaped
                "in": sorted(missing_user_ids)
            }
        }
    )
    email_by_user_id: Final = MappingProxyType({u.user_id: u.user_email for u in user_rows if u.user_email})

    return tuple(
        m.model_copy(update={"user_email": email_by_user_id[m.user_id]})  # mutable-ok: pydantic update payload
        if not m.user_email and m.user_id is not None and m.user_id in email_by_user_id
        else m
        for m in members
    )


async def _resolve_team_access_group_resources(
    _team_info: TeamInfoResponseObjectTeamTable,
) -> TeamInfoResponseObjectTeamTable:
    """Return a copy of the team info with access_group_models / mcp_server_ids /
    agent_ids / details resolved from its access groups."""
    if not _team_info.access_group_ids:
        return _team_info
    ag_lookup: Final = await _batch_resolve_access_group_resources(_team_info.access_group_ids)
    resolved_groups: Final = tuple(
        ag_lookup[ag_id] for ag_id in dict.fromkeys(_team_info.access_group_ids) if ag_id in ag_lookup
    )
    return _team_info.model_copy(
        update={
            "access_group_models": list({m for group in resolved_groups for m in (group.access_model_names or [])}),
            "access_group_mcp_server_ids": list(
                {s for group in resolved_groups for s in (group.access_mcp_server_ids or [])}
            ),
            "access_group_agent_ids": list({a for group in resolved_groups for a in (group.access_agent_ids or [])}),
            "access_group_details": tuple(
                TeamAccessGroupModelGrant(
                    access_group_id=group.access_group_id,
                    access_group_name=group.access_group_name,
                    models=tuple(group.access_model_names or ()),
                )
                for group in resolved_groups
            ),
        }
    )


@router.get("/team/info", tags=["team management"], dependencies=[Depends(user_api_key_auth)])
@management_endpoint_wrapper
async def team_info(
    http_request: Request,
    team_id: str = fastapi.Query(default=None, description="Team ID in the request parameters"),
    key_limit: int | None = fastapi.Query(default=None, description="Limit the number of keys returned", gt=0),
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    get info on team + related keys

    Parameters:
    - team_id: str - Required. The unique identifier of the team to get info on.

    ```
    curl --location 'http://localhost:4000/team/info?team_id=your_team_id_here' \
    --header 'Authorization: Bearer your_api_key_here'
    ```
    """
    from litellm.proxy._types import TeamInfoResponseObjectTeamTable
    from litellm.proxy.proxy_server import prisma_client

    try:
        if prisma_client is None:
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail={
                    "error": "Database not connected. Connect a database to your proxy - https://docs.litellm.ai/docs/simple_proxy#managing-auth---virtual-keys"
                },
            )
        if team_id is None:
            raise HTTPException(
                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
                detail={"message": "Malformed request. No team id passed in."},
            )

        try:
            team_info: BaseModel | None = await _team_db(prisma_client).find_unique(
                where={"team_id": team_id},
                include={"litellm_model_table": True, "object_permission": True},
            )
            if team_info is None:
                raise Exception
        except Exception:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail={"message": f"Team not found, passed team id: {team_id}."},
            )
        await validate_membership(
            user_api_key_dict=user_api_key_dict,
            team_table=LiteLLM_TeamTable.model_validate(team_info.model_dump()),
        )

        ## GET ALL KEYS ##
        keys = await prisma_client.get_data(
            team_id=team_id,
            table_name="key",
            query_type="find_all",
            expires=datetime.now(),
            limit=key_limit,
        )

        if keys is None:
            keys = []

        if team_info is None:
            ## make sure we still return a total spend ##
            spend = 0
            for k in keys:
                spend += getattr(k, "spend", 0)
            team_info = {"spend": spend}

        ## REMOVE HASHED TOKEN INFO before returning ##
        for key in keys:
            try:
                key = key.model_dump()
            except Exception:
                # if using pydantic v1
                key = key.dict()
            key.pop("token", None)

        ## GET ALL MEMBERSHIPS ##
        returned_tm: Final = await get_all_team_memberships(prisma_client, [team_id], user_id=None)

        if isinstance(team_info, dict):
            _team_info = TeamInfoResponseObjectTeamTable.model_validate(team_info)
        elif isinstance(team_info, BaseModel):
            _team_info = TeamInfoResponseObjectTeamTable.model_validate(team_info.model_dump())
        else:
            _team_info = TeamInfoResponseObjectTeamTable()

        ## GET TEAM BUDGET (if exists) ##
        team_member_budget_id: Final = (
            _team_info.metadata.get("team_member_budget_id") if _team_info.metadata is not None else None
        )
        if team_member_budget_id is not None:
            _team_info = await _add_team_member_budget_table(
                team_member_budget_id=team_member_budget_id,
                prisma_client=prisma_client,
                team_info_response_object=_team_info,
            )

        # Resolve resources inherited from access groups
        resolved_team_info: Final = await _resolve_team_access_group_resources(_team_info)

        # Fill in emails the add-time roster snapshot never captured
        hydrated_members: Final = await _hydrate_member_emails(
            prisma_client=prisma_client,
            members=resolved_team_info.members_with_roles,
        )
        hydrated_team_info: Final = resolved_team_info.model_copy(
            update={  # mutable-ok: pydantic update payload
                # list(), not the tuple: model_copy skips validation, so the field has
                # to be handed the list[Member] the response model declares.
                "members_with_roles": list(hydrated_members)  # mutable-ok: declared list[Member]
            }
        )

        response_object: Final = TeamInfoResponseObject(
            team_id=team_id,
            team_info=hydrated_team_info,
            keys=keys,
            team_memberships=returned_tm,
        )
        return response_object

    except Exception as e:
        verbose_proxy_logger.error(
            "litellm.proxy.management_endpoints.team_endpoints.py::team_info - Exception occurred - %s\n%s",
            e,
            traceback.format_exc(),
        )
        if isinstance(e, HTTPException):
            raise ProxyException(
                message=getattr(e, "detail", f"Authentication Error({e})"),
                type=ProxyErrorTypes.auth_error,
                param=getattr(e, "param", "None"),
                code=getattr(e, "status_code", status.HTTP_400_BAD_REQUEST),
            )
        elif isinstance(e, ProxyException):
            raise e
        raise ProxyException(
            message="Authentication Error, " + str(e),
            type=ProxyErrorTypes.auth_error,
            param=getattr(e, "param", "None"),
            code=status.HTTP_400_BAD_REQUEST,
        )


@router.get(
    "/team/{team_id}/members/me",
    tags=["team management"],
    dependencies=[Depends(user_api_key_auth)],
    response_model=TeamMemberInfoResponse,
)
@management_endpoint_wrapper
async def team_member_me(
    http_request: Request,
    team_id: str,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    Get the caller's own team-membership row for the given team.

    Used by internal users to view their own spend, budget, budget reset
    date, rate limits, and role within a team — without exposing other
    members' data. The caller is resolved from their API key; the path
    `/members/me` always refers to that caller.

    Returns 404 if the caller is not a member of the team.

    ```
    curl --location 'http://localhost:4000/team/your_team_id/members/me' \
    --header 'Authorization: Bearer your_api_key_here'
    ```
    """
    from litellm.proxy.proxy_server import prisma_client, user_api_key_cache

    if prisma_client is None:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail={
                "error": "Database not connected. Connect a database to your proxy - https://docs.litellm.ai/docs/simple_proxy#managing-auth---virtual-keys"
            },
        )

    caller_user_id: Final = user_api_key_dict.user_id
    if caller_user_id is None:
        # Team keys / service-account keys without a user_id can't resolve "me".
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail={"error": "API key has no associated user_id; cannot resolve 'me' for team membership."},
        )

    team_table: Final = await get_team_object(
        team_id=team_id,
        prisma_client=prisma_client,
        user_api_key_cache=user_api_key_cache,
    )

    caller_user_email: Final = user_api_key_dict.user_email
    member_role: str | None = None
    for m in team_table.members_with_roles:
        # Match by user_id when present, else fall back to email — members
        # added by email may have user_id=None on the stored entry.
        if (m.user_id is not None and m.user_id == caller_user_id) or (
            m.user_email is not None and caller_user_email is not None and m.user_email == caller_user_email
        ):
            member_role = m.role
            break

    if member_role is None:
        # Caller is not a member of this team. Even proxy admins get 404 here —
        # they can use /team/info to view all members; "me" only resolves for
        # actual members of the team.
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail={"error": f"User user_id={caller_user_id} is not a member of team_id={team_id}."},
        )

    membership: Final = await get_team_membership(
        user_id=caller_user_id,
        team_id=team_id,
        prisma_client=prisma_client,
        user_api_key_cache=user_api_key_cache,
    )

    user_row: Final = await get_user_object(
        user_id=caller_user_id,
        prisma_client=prisma_client,
        user_api_key_cache=user_api_key_cache,
        user_id_upsert=False,
    )
    user_email: Final = getattr(user_row, "user_email", None) if user_row is not None else None

    if membership is None:
        # Member is in members_with_roles but has no membership row yet
        # (no per-member budget/limits configured). Return defaults.
        return TeamMemberInfoResponse(
            user_id=caller_user_id,
            team_id=team_id,
            team_alias=team_table.team_alias,
            role=member_role,
            user_email=user_email,
            spend=0.0,
            total_spend=0.0,
            budget_id=None,
            litellm_budget_table=None,
        )

    return TeamMemberInfoResponse(
        user_id=caller_user_id,
        team_id=team_id,
        team_alias=team_table.team_alias,
        role=member_role,
        user_email=user_email,
        spend=membership.spend,
        total_spend=membership.total_spend,
        budget_id=membership.budget_id,
        litellm_budget_table=membership.litellm_budget_table,
    )


@router.post("/team/block", tags=["team management"], dependencies=[Depends(user_api_key_auth)])
@management_endpoint_wrapper
async def block_team(
    data: BlockTeamRequest,
    http_request: Request,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    Blocks all calls from keys with this team id.

    Parameters:
    - team_id: str - Required. The unique identifier of the team to block.

    Example:
    ```
    curl --location 'http://0.0.0.0:4000/team/block' \
    --header 'Authorization: Bearer sk-1234' \
    --header 'Content-Type: application/json' \
    --data '{
        "team_id": "team-1234"
    }'
    ```

    Returns:
    - The updated team record with blocked=True



    """
    from litellm.proxy.proxy_server import prisma_client

    if prisma_client is None:
        raise Exception("No DB Connected.")

    existing_team: Final = await _team_db(prisma_client).find_unique(where={"team_id": data.team_id})
    if existing_team is None:
        raise HTTPException(
            status_code=404,
            detail={"error": f"Team not found, passed team_id={data.team_id}"},
        )

    # Verify caller has access to manage this team
    await _verify_team_access(
        team_obj=LiteLLM_TeamTable.model_validate(existing_team.model_dump()),
        user_api_key_dict=user_api_key_dict,
    )

    record: Final = await _team_db(prisma_client).update(
        where={"team_id": data.team_id},
        data={"blocked": True},
    )

    return record


@router.post("/team/unblock", tags=["team management"], dependencies=[Depends(user_api_key_auth)])
@management_endpoint_wrapper
async def unblock_team(
    data: BlockTeamRequest,
    http_request: Request,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    Unblocks a previously blocked team, re-enabling calls from keys with this team id.

    Parameters:
    - team_id: str - Required. The unique identifier of the team to unblock.

    Example:
    ```
    curl --location 'http://0.0.0.0:4000/team/unblock' \
    --header 'Authorization: Bearer sk-1234' \
    --header 'Content-Type: application/json' \
    --data '{
        "team_id": "team-1234"
    }'
    ```
    """
    from litellm.proxy.proxy_server import prisma_client

    if prisma_client is None:
        raise Exception("No DB Connected.")

    existing_team: Final = await _team_db(prisma_client).find_unique(where={"team_id": data.team_id})
    if existing_team is None:
        raise HTTPException(
            status_code=404,
            detail={"error": f"Team not found, passed team_id={data.team_id}"},
        )

    # Verify caller has access to manage this team
    await _verify_team_access(
        team_obj=LiteLLM_TeamTable.model_validate(existing_team.model_dump()),
        user_api_key_dict=user_api_key_dict,
    )

    record: Final = await _team_db(prisma_client).update(
        where={"team_id": data.team_id},
        data={"blocked": False},
    )

    return record


@router.get(
    "/team/metadata_schema",
    tags=["team management"],  # mutable-ok: fastapi's decorator signature types tags as a list
    dependencies=(Depends(user_api_key_auth),),
    response_model=TeamMetadataSchemaResponse,
)
async def get_team_metadata_schema():
    """
    Get the team metadata fields declared in ``general_settings.team_metadata_schema``.

    The UI uses this to prepopulate the team metadata form with the declared
    keys. Returns an empty ``fields`` list when no schema is configured. This
    schema is advisory; server-side enforcement stays with
    ``custom_team_metadata_validate``.
    """
    return TeamMetadataSchemaResponse(fields=TEAM_METADATA_SCHEMA_REGISTRY.get())


@router.get("/team/available")
async def list_available_teams(
    http_request: Request,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
    response_model=list[LiteLLM_TeamTable],
):
    from litellm.proxy.proxy_server import prisma_client

    if prisma_client is None:
        raise HTTPException(
            status_code=400,
            detail={"error": CommonProxyErrors.db_not_connected_error.value},
        )

    available_teams = cast(
        list[str] | None,
        (
            litellm.default_internal_user_params.get("available_teams")
            if litellm.default_internal_user_params is not None
            else None
        ),
    )
    if available_teams is None:
        return []

    # filter out teams that the user is already a member of
    user_info: Final = await _user_db(prisma_client).find_unique(where={"user_id": user_api_key_dict.user_id})
    if user_info is None:
        raise HTTPException(
            status_code=404,
            detail={"error": "User not found"},
        )
    user_info_correct_type: Final = LiteLLM_UserTable.model_validate(user_info.model_dump())

    available_teams = [team for team in available_teams if team not in user_info_correct_type.teams]

    available_teams_db: Final = await _team_db(prisma_client).find_many(where={"team_id": {"in": available_teams}})

    available_teams_correct_type = [LiteLLM_TeamTable.model_validate(team.model_dump()) for team in available_teams_db]

    return available_teams_correct_type


async def _get_org_admin_org_ids(
    user_id: str,
    prisma_client: PrismaClient,
    user_api_key_cache: UserApiKeyCache,
    proxy_logging_obj: ProxyLogging,
) -> list[str] | None:
    """
    Return the list of organization IDs where the user is an org admin.
    Returns None if the user is not an org admin of any organization or if
    the user cannot be found.
    """
    try:
        caller_user: Final = await get_user_object(
            user_id=user_id,
            prisma_client=prisma_client,
            user_api_key_cache=user_api_key_cache,
            user_id_upsert=False,
            proxy_logging_obj=proxy_logging_obj,
        )
    except ValueError:
        # get_user_object raises ValueError when the user doesn't exist
        return None

    if caller_user is None:
        return None

    org_ids: Final = [
        m.organization_id
        for m in (caller_user.organization_memberships or [])
        if m.user_role == LitellmUserRoles.ORG_ADMIN.value and m.organization_id is not None
    ]
    return org_ids if org_ids else None


async def _build_team_list_where_conditions(
    prisma_client: PrismaClient,
    team_id: str | None,
    team_alias: str | None,
    organization_id: str | None,
    user_id: str | None,
    use_deleted_table: bool,
    search: str | None = None,
    search_team_id_match: TeamIdSearchMatch = "exact",
    org_admin_org_ids: list[str] | None = None,
    user_api_key_cache: UserApiKeyCache | None = None,
    proxy_logging_obj: ProxyLogging | None = None,
) -> dict[str, object] | None:
    """
    Build where conditions for team list query.

    Returns None when the query is guaranteed to yield no results (e.g. user
    has no team memberships), allowing the caller to skip the DB round-trip.
    """
    where_conditions: Final[dict[str, object]] = {}

    if team_id:
        where_conditions["team_id"] = team_id

    if team_alias:
        where_conditions["team_alias"] = {
            "contains": team_alias,
            "mode": "insensitive",  # Case-insensitive search
        }

    if search:
        where_conditions["OR"] = [
            ({"team_id": {"startsWith": search}} if search_team_id_match == "prefix" else {"team_id": search}),
            {"team_alias": {"contains": search, "mode": "insensitive"}},
        ]

    if organization_id:
        where_conditions["organization_id"] = organization_id
    elif org_admin_org_ids is not None:
        # Org admin: always scope to their orgs, even when filtering by user_id.
        where_conditions["organization_id"] = {"in": org_admin_org_ids}

    if user_id:
        try:
            user_object_correct_type: Final = await get_user_object(
                user_id=user_id,
                prisma_client=prisma_client,
                user_api_key_cache=user_api_key_cache,
                user_id_upsert=False,
                proxy_logging_obj=proxy_logging_obj,
            )
        except ValueError:
            raise HTTPException(
                status_code=404,
                detail={"error": f"User not found, passed user_id={user_id}"},
            )
        if user_object_correct_type is None:
            raise HTTPException(
                status_code=404,
                detail={"error": f"User not found, passed user_id={user_id}"},
            )
        user_team_ids: Final = user_object_correct_type.teams or []

        if use_deleted_table:
            where_conditions["members"] = {"has": user_id}
        else:
            # When user_id is provided, filter by that user's direct team
            # memberships. For org admins the access control gate in
            # list_team_v2 already verified the caller's authority — the
            # filter logic is the same as for regular users.
            if not user_team_ids:
                return None  # no memberships — skip the DB query
            elif team_id is not None:
                # team_id exact-match already in where_conditions; verify membership
                if team_id not in user_team_ids:
                    raise HTTPException(
                        status_code=404,
                        detail={"error": f"User is not a member of team_id={team_id}"},
                    )
            else:
                where_conditions["team_id"] = {"in": user_team_ids}

    return where_conditions


async def _batch_resolve_access_group_resources(
    all_access_group_ids: list[str],
) -> "dict[str, prisma_models.LiteLLM_AccessGroupTable]":
    """
    Batch-fetch access groups in a single DB query and return them keyed by
    access_group_id. Missing/invalid groups are silently omitted.
    """
    from litellm.proxy.proxy_server import prisma_client as _prisma_client

    if not all_access_group_ids or _prisma_client is None:
        return {}

    unique_ids: Final = tuple(frozenset(all_access_group_ids))
    rows: Final = await _access_group_db(_prisma_client).find_many(
        where={"access_group_id": {"in": unique_ids}},
    )
    return {row.access_group_id: row for row in rows}


def _convert_teams_to_response_models(
    teams: Sequence,
    use_deleted_table: bool,
    keys_count_by_team: dict[str, int] | None = None,
) -> list[TeamListItem | LiteLLM_TeamTable | LiteLLM_DeletedTeamTable]:
    """Convert raw Prisma team rows to response models."""
    team_list: Final[list[TeamListItem | LiteLLM_TeamTable | LiteLLM_DeletedTeamTable]] = []
    counts: Final = keys_count_by_team or {}
    for team in teams:
        try:
            team_dict = team.model_dump()
        except Exception:
            team_dict = team.dict()

        if use_deleted_table:
            team_list.append(LiteLLM_DeletedTeamTable.model_validate(team_dict))
        else:
            members_with_roles = team_dict.get("members_with_roles")
            if not isinstance(members_with_roles, list):
                members_with_roles = []
                team_dict["members_with_roles"] = members_with_roles
            members_count = len(members_with_roles)
            keys_count = counts.get(team_dict.get("team_id") or "", 0)
            team_list.append(
                TeamListItem(
                    **team_dict,
                    members_count=members_count,
                    keys_count=keys_count,
                )
            )
    return team_list


async def _get_keys_count_by_team(
    prisma_client: PrismaClient,
    teams: Sequence[_TeamIdRow],
) -> dict[str, int]:
    """Aggregate virtual-key counts per team for the given page of teams.

    Runs a single GROUP BY against LiteLLM_VerificationToken. The IN clause is
    bounded by page_size and uses the existing @@index([team_id]), so this is
    one DB round-trip per page. Returns an empty map when the page has no teams.
    """
    page_team_ids: Final = [getattr(t, "team_id", None) for t in teams if getattr(t, "team_id", None)]
    if not page_team_ids:
        return {}

    grouped: Final = cast(  # cast-ok: prisma group_by returns one row per `by` key with `count=` nested under "_count"
        "Sequence[_TeamIdGroupRow]",
        await _tokens_db(prisma_client).group_by(
            by=["team_id"],
            where={"team_id": {"in": page_team_ids}},
            count={"team_id": True},
        ),
    )
    return {row["team_id"]: row.get("_count", {}).get("team_id", 0) for row in grouped if row.get("team_id")}


async def _enforce_list_team_v2_access(
    user_api_key_dict: UserAPIKeyAuth,
    user_id: str | None,
    organization_id: str | None,
    prisma_client: PrismaClient,
    user_api_key_cache: UserApiKeyCache,
    proxy_logging_obj: ProxyLogging,
) -> tuple[str | None, list[str] | None]:
    """Enforce access control for list_team_v2.

    - Proxy admins and admin viewers can query any teams.
    - Org admins can query teams within their organizations.
    - Regular users can only query their own teams.

    Returns the (possibly overridden) user_id and org_admin_org_ids.
    """
    is_proxy_admin: Final = _user_has_admin_view(user_api_key_dict)
    org_admin_org_ids: list[str] | None = None

    if is_proxy_admin:
        return user_id, org_admin_org_ids

    # Always check org admin status so that even own-queries see
    # the full set of organisation teams, not just direct memberships.
    if user_api_key_dict.user_id:
        org_admin_org_ids = await _get_org_admin_org_ids(
            user_id=user_api_key_dict.user_id,
            prisma_client=prisma_client,
            user_api_key_cache=user_api_key_cache,
            proxy_logging_obj=proxy_logging_obj,
        )

    if org_admin_org_ids is not None:
        # Org admin: validate org_id filter if provided
        if organization_id and organization_id not in org_admin_org_ids:
            raise HTTPException(
                status_code=403,
                detail={"error": "You can only view teams within your organizations."},
            )
        # When the caller is an org admin querying their own teams (or no
        # specific user), null out user_id so that
        # _build_team_list_where_conditions scopes only by organization_id
        # — org admins should see all teams in their orgs, not just teams
        # they are a direct member of.  Keep user_id when the org admin
        # explicitly queries a *different* user's teams.
        if user_id is None or user_id == user_api_key_dict.user_id:
            user_id = None
        verbose_proxy_logger.debug(
            "list_team_v2: org admin access for user=%s, org_ids=%s, user_id_filter=%s",
            user_api_key_dict.user_id,
            org_admin_org_ids,
            user_id,
        )
    else:
        # Not an org admin — fall back to standard route check
        if not allowed_route_check_inside_route(user_api_key_dict=user_api_key_dict, requested_user_id=user_id):
            raise HTTPException(
                status_code=401,
                detail={
                    "error": f"Only admin users can query all teams/other teams. Your user role={user_api_key_dict.user_role}"
                },
            )
        # Regular user — auto-inject caller's user_id
        if user_id is None:
            user_id = user_api_key_dict.user_id

    return user_id, org_admin_org_ids


@router.get(
    "/v2/team/list",
    tags=["team management"],
    response_model=TeamListResponse,
    dependencies=[Depends(user_api_key_auth)],
)
@management_endpoint_wrapper
async def list_team_v2(
    http_request: Request,
    user_id: str | None = fastapi.Query(default=None, description="Only return teams which this 'user_id' belongs to"),
    organization_id: str | None = fastapi.Query(
        default=None,
        description="Only return teams which this 'organization_id' belongs to",
    ),
    team_id: str | None = fastapi.Query(default=None, description="Only return teams which this 'team_id' belongs to"),
    team_alias: str | None = fastapi.Query(
        default=None,
        description="Only return teams which this 'team_alias' belongs to. Supports partial matching.",
    ),
    search: str | None = fastapi.Query(
        default=None,
        description="Combined search: matches teams whose 'team_id' matches the value OR whose 'team_alias' contains it (case-insensitive).",
    ),
    search_team_id_match: Annotated[
        TeamIdSearchMatch,
        fastapi.Query(
            description="How 'search' matches 'team_id': 'exact' (default) or 'prefix' for a case-sensitive prefix match."
        ),
    ] = "exact",
    page: int = fastapi.Query(default=1, description="Page number for pagination", ge=1),
    page_size: int = fastapi.Query(default=10, description="Number of teams per page", ge=1, le=100),
    sort_by: str | None = fastapi.Query(
        default=None,
        description="Column to sort by (e.g. 'team_id', 'team_alias', 'created_at')",
    ),
    sort_order: str = fastapi.Query(default="asc", description="Sort order ('asc' or 'desc')"),
    status: str | None = fastapi.Query(default=None, description="Filter by status (e.g. 'deleted')"),
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    Get a paginated list of teams with filtering and sorting options.

    Parameters:
        user_id: Optional[str]
            Only return teams which this user belongs to
        organization_id: Optional[str]
            Only return teams which belong to this organization
        team_id: Optional[str]
            Filter teams by exact team_id match
        team_alias: Optional[str]
            Filter teams by partial team_alias match
        page: int
            The page number to return
        page_size: int
            The number of items per page
        sort_by: Optional[str]
            Column to sort by (e.g. 'team_id', 'team_alias', 'created_at')
        sort_order: str
            Sort order ('asc' or 'desc')
        status: Optional[str]
            Filter by status. Currently supports "deleted" to query deleted teams.
    """
    from litellm.proxy.proxy_server import (
        prisma_client,
        proxy_logging_obj,
        user_api_key_cache,
    )

    if prisma_client is None:
        raise HTTPException(
            status_code=500,
            detail={"error": f"No db connected. prisma client={prisma_client}"},
        )

    # --- Access control ---
    user_id, org_admin_org_ids = await _enforce_list_team_v2_access(
        user_api_key_dict=user_api_key_dict,
        user_id=user_id,
        organization_id=organization_id,
        prisma_client=prisma_client,
        user_api_key_cache=user_api_key_cache,
        proxy_logging_obj=proxy_logging_obj,
    )

    if status is not None and status != "deleted":
        raise HTTPException(
            status_code=400,
            detail={"error": "Invalid status value. Currently only 'deleted' is supported."},
        )

    use_deleted_table: Final = status == "deleted"

    # Calculate skip and take for pagination
    skip: Final = (page - 1) * page_size

    # Build where conditions based on provided parameters.
    # Returns None when the query is guaranteed to yield no results.
    where_conditions: Final = await _build_team_list_where_conditions(
        prisma_client=prisma_client,
        team_id=team_id,
        team_alias=team_alias,
        organization_id=organization_id,
        user_id=user_id,
        use_deleted_table=use_deleted_table,
        search=search,
        search_team_id_match=search_team_id_match,
        org_admin_org_ids=org_admin_org_ids,
        user_api_key_cache=user_api_key_cache,
        proxy_logging_obj=proxy_logging_obj,
    )

    if where_conditions is None:
        return {
            "teams": [],
            "total": 0,
            "page": page,
            "page_size": page_size,
            "total_pages": 0,
        }

    # Build order_by conditions
    valid_sort_columns: Final = ["team_id", "team_alias", "created_at"]
    order_by = None
    if sort_by and sort_by in valid_sort_columns:
        if sort_order.lower() not in ["asc", "desc"]:
            sort_order = "asc"
        order_by = {sort_by: sort_order.lower()}

    # Get teams with pagination
    if use_deleted_table:
        # LiteLLM_DeletedTeamTable has no litellm_model_table relation, unlike below
        teams = await _deleted_team_db(prisma_client).find_many(
            where=where_conditions,
            skip=skip,
            take=page_size,
            order=order_by if order_by else {"created_at": "desc"},  # Default sort
        )
        # Get total count for pagination
        total_count = await _deleted_team_db(prisma_client).count(where=where_conditions)
    else:
        teams = await _team_db(prisma_client).find_many(
            where=where_conditions,
            skip=skip,
            take=page_size,
            order=order_by if order_by else {"created_at": "desc"},  # Default sort
            include=_INCLUDE_MODEL_TABLE,
        )
        # Get total count for pagination
        total_count = await _team_db(prisma_client).count(where=where_conditions)

    # Calculate total pages
    total_pages: Final = -(-total_count // page_size)  # Ceiling division

    # Aggregate virtual-key counts per team for the current page. The deleted
    # table does not carry keys_count, so it is skipped.
    keys_count_by_team: dict[str, int] = {}
    if not use_deleted_table:
        keys_count_by_team = await _get_keys_count_by_team(prisma_client, teams)

    # Convert Prisma models to response models with members_count and keys_count
    team_list = _convert_teams_to_response_models(teams, use_deleted_table, keys_count_by_team=keys_count_by_team)

    # Resolve resources inherited from access groups (single batch query)
    if not use_deleted_table:
        team_items_with_ag: Final = [t for t in team_list if isinstance(t, TeamListItem) and t.access_group_ids]
        if team_items_with_ag:
            all_ag_ids: Final = [ag_id for t in team_items_with_ag for ag_id in (t.access_group_ids or [])]
            ag_lookup: Final = await _batch_resolve_access_group_resources(all_ag_ids)
            for team_item in team_items_with_ag:
                team_groups = tuple(
                    ag_lookup[ag_id] for ag_id in (team_item.access_group_ids or []) if ag_id in ag_lookup
                )
                team_item.access_group_models = list(
                    {m for group in team_groups for m in (group.access_model_names or [])}
                )
                team_item.access_group_mcp_server_ids = list(
                    {s for group in team_groups for s in (group.access_mcp_server_ids or [])}
                )
                team_item.access_group_agent_ids = list(
                    {a for group in team_groups for a in (group.access_agent_ids or [])}
                )

    return {
        "teams": team_list,
        "total": total_count,
        "page": page,
        "page_size": page_size,
        "total_pages": total_pages,
    }


async def _authorize_and_filter_teams(
    user_api_key_dict: UserAPIKeyAuth,
    user_id: str | None,
    prisma_client: PrismaClient,
    user_api_key_cache: UserApiKeyCache,
    proxy_logging_obj: ProxyLogging,
) -> list:
    """
    Authorize the /team/list request and return filtered teams.

    - Proxy admins: all teams (or filtered by user_id if provided).
    - Org admins: teams from their orgs (scoped to user_id if provided).
    - Own query (user_id matches caller): teams the user is a member of.
    - Others: 401.
    """
    is_proxy_admin: Final = _user_has_admin_view(user_api_key_dict)
    allowed_org_ids: list[str] | None = None

    if not is_proxy_admin:
        is_own_query: Final = (
            user_id is not None and user_api_key_dict.user_id is not None and user_api_key_dict.user_id == user_id
        )

        # Check if user is an org admin (even for own queries, so they see org teams)
        if user_api_key_dict.user_id is not None:
            caller_user: Final = await get_user_object(
                user_id=user_api_key_dict.user_id,
                prisma_client=prisma_client,
                user_api_key_cache=user_api_key_cache,
                user_id_upsert=False,
                proxy_logging_obj=proxy_logging_obj,
            )
            if caller_user is not None:
                allowed_org_ids = [
                    m.organization_id
                    for m in (caller_user.organization_memberships or [])
                    if m.user_role == LitellmUserRoles.ORG_ADMIN.value and m.organization_id is not None
                ]
                if not allowed_org_ids:
                    allowed_org_ids = None

        if allowed_org_ids is None and not is_own_query:
            raise HTTPException(
                status_code=401,
                detail={
                    "error": f"Only admin users can query all teams/other teams. Your user role={user_api_key_dict.user_role}"
                },
            )

    if allowed_org_ids is not None:
        # Org admin: query DB for teams in their orgs
        org_teams: Final = await _raw_team_db(TeamRepository(prisma_client)).find_many(
            where={"organization_id": {"in": allowed_org_ids}},
            include={"litellm_model_table": True},
        )
        if not user_id:
            return list(org_teams)
        # Filter org teams to only those where the target user is a member
        return [
            team
            for team in org_teams
            if team.members_with_roles and any(m.get("user_id") == user_id for m in team.members_with_roles)
        ]
    elif user_id:
        # Regular user: fetch all and filter by membership (Prisma can't filter JSON arrays)
        response: Final = await _raw_team_db(TeamRepository(prisma_client)).find_many(
            include={"litellm_model_table": True}
        )
        return [
            team
            for team in response
            if team.members_with_roles and any(m.get("user_id") == user_id for m in team.members_with_roles)
        ]
    else:
        # Proxy admin: all teams
        return list(await _raw_team_db(TeamRepository(prisma_client)).find_many(include={"litellm_model_table": True}))


@router.get("/team/list", tags=["team management"], dependencies=[Depends(user_api_key_auth)])
@management_endpoint_wrapper
async def list_team(
    http_request: Request,
    user_id: str | None = fastapi.Query(default=None, description="Only return teams which this 'user_id' belongs to"),
    organization_id: str | None = None,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    ```
    curl --location --request GET 'http://0.0.0.0:4000/team/list' \
        --header 'Authorization: Bearer sk-1234'
    ```

    Parameters:
    - user_id: str - Optional. If passed will only return teams that the user_id is a member of.
    - organization_id: str - Optional. If passed will only return teams that belong to the organization_id. Pass 'default_organization' to get all teams without organization_id.
    """
    from litellm.proxy.proxy_server import (
        prisma_client,
        proxy_logging_obj,
        user_api_key_cache,
    )

    if prisma_client is None:
        raise HTTPException(
            status_code=400,
            detail={"error": CommonProxyErrors.db_not_connected_error.value},
        )

    filtered_response: Final = await _authorize_and_filter_teams(
        user_api_key_dict=user_api_key_dict,
        user_id=user_id,
        prisma_client=prisma_client,
        user_api_key_cache=user_api_key_cache,
        proxy_logging_obj=proxy_logging_obj,
    )

    _team_ids: Final = [team.team_id for team in filtered_response]
    returned_tm: Final = await get_all_team_memberships(prisma_client, _team_ids, user_id=user_id)

    returned_responses: list[TeamListResponseObject] = []
    for team in filtered_response:
        _team_memberships: list[LiteLLM_TeamMembership] = []
        for tm in returned_tm:
            if tm.team_id == team.team_id:
                _team_memberships.append(tm)

        # add all keys that belong to the team
        keys = _as_list(await _tokens_db(prisma_client).find_many(where={"team_id": team.team_id}))

        try:
            returned_responses.append(
                TeamListResponseObject(
                    **team.model_dump(),
                    team_memberships=_team_memberships,
                    keys=keys,
                )
            )
        except Exception as e:
            team_exception = f"""Invalid team object for team_id: {team.team_id}. team_object={team.model_dump()}.
            Error: {e}
            """
            verbose_proxy_logger.exception(team_exception)
            continue
    # Sort the responses by team_alias
    returned_responses.sort(key=lambda x: getattr(x, "team_alias", "") or "")

    if organization_id is not None:
        if organization_id == SpecialManagementEndpointEnums.DEFAULT_ORGANIZATION.value:
            returned_responses = [team for team in returned_responses if team.organization_id is None]
        else:
            returned_responses = [team for team in returned_responses if team.organization_id == organization_id]

    return returned_responses


async def get_paginated_teams(
    prisma_client: PrismaClient,
    page_size: int = 10,
    page: int = 1,
) -> tuple[list[LiteLLM_TeamTable], int]:
    """
    Get paginated list of teams from team table

    Parameters:
        prisma_client: PrismaClient - The database client
        page_size: int - Number of teams per page
        page: int - Page number (1-based)

    Returns:
        Tuple[List[LiteLLM_TeamTable], int] - (list of teams, total count)
    """
    try:
        # Calculate skip for pagination
        skip: Final = (page - 1) * page_size
        # Get total count
        total_count: Final = await _team_db(prisma_client).count()

        # Get paginated teams
        teams: Final = await _team_db(prisma_client).find_many(
            skip=skip,
            take=page_size,
            order={"team_alias": "asc"},  # Sort by team_alias
        )
        return teams, total_count
    except Exception as e:
        verbose_proxy_logger.exception("[Non-Blocking] Error getting paginated teams: %s", e)
        return [], 0


@router.get(
    "/team/filter/ui",
    tags=["team management"],
    dependencies=[Depends(user_api_key_auth)],
    include_in_schema=False,
    responses={
        200: {"model": list[LiteLLM_TeamTable]},
    },
)
async def ui_view_teams(
    team_id: str | None = fastapi.Query(default=None, description="Team ID in the request parameters"),
    team_alias: str | None = fastapi.Query(default=None, description="Team alias in the request parameters"),
    page: int = fastapi.Query(default=1, description="Page number for pagination", ge=1),
    page_size: int = fastapi.Query(default=50, description="Number of items per page", ge=1, le=100),
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    [PROXY-ADMIN ONLY] Filter teams based on partial match of team_id or team_alias with pagination.

    Args:
        user_id (Optional[str]): Partial user ID to search for
        user_email (Optional[str]): Partial email to search for
        page (int): Page number for pagination (starts at 1)
        page_size (int): Number of items per page (max 100)
        user_api_key_dict (UserAPIKeyAuth): User authentication information

    Returns:
        List[LiteLLM_SpendLogs]: Paginated list of matching user records
    """
    from litellm.proxy.proxy_server import prisma_client

    if prisma_client is None:
        raise HTTPException(status_code=500, detail={"error": "No db connected"})

    try:
        # Calculate offset for pagination
        skip: Final = (page - 1) * page_size

        # Build where conditions based on provided parameters
        where_conditions: Final[_TeamUiViewFilters] = {}

        if team_id:
            where_conditions["team_id"] = {
                "contains": team_id,
                "mode": "insensitive",  # Case-insensitive search
            }

        if team_alias:
            where_conditions["team_alias"] = {
                "contains": team_alias,
                "mode": "insensitive",  # Case-insensitive search
            }

        # Query users with pagination and filters
        teams: Final = await _team_db(prisma_client).find_many(
            where=where_conditions,
            skip=skip,
            take=page_size,
            order={"created_at": "desc"},
        )

        if not teams:
            return []

        return teams

    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error searching teams: {e}")


def add_new_models_to_team(team_obj: LiteLLM_TeamTable, new_models: list[str]) -> list[str]:
    """
    Add new models to a team's allowed model list.
    """
    current_models = team_obj.models
    if current_models is not None and len(current_models) == 0:  # implies all model access
        current_models = [SpecialModelNames.all_proxy_models.value]
    else:
        current_models = team_obj.models
    updated_models: Final = list(set(current_models + new_models))
    return updated_models


@router.post(
    "/team/model/add",
    tags=["team management"],
    dependencies=[Depends(user_api_key_auth)],
)
@management_endpoint_wrapper
async def team_model_add(
    data: TeamModelAddRequest,
    http_request: Request,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    Add models to a team's allowed model list. Only proxy admin or team admin can add models.

    Parameters:
    - team_id: str - Required. The team to add models to
    - models: List[str] - Required. List of models to add to the team

    Example Request:
    ```
    curl --location 'http://0.0.0.0:4000/team/model/add' \
    --header 'Authorization: Bearer sk-1234' \
    --header 'Content-Type: application/json' \
    --data '{
        "team_id": "team-1234",
        "models": ["gpt-4", "claude-2"]
    }'
    ```
    """
    from litellm.proxy.proxy_server import (
        prisma_client,
        proxy_logging_obj,
        user_api_key_cache,
    )

    if prisma_client is None:
        raise HTTPException(status_code=500, detail={"error": "No db connected"})

    # Get existing team
    team_row: Final = await _team_db(prisma_client).find_unique(where={"team_id": data.team_id})

    if team_row is None:
        raise HTTPException(
            status_code=404,
            detail={"error": f"Team not found, passed team_id={data.team_id}"},
        )

    team_obj: Final = LiteLLM_TeamTable.model_validate(team_row.model_dump())

    # Authorization check - only proxy admin, team admin, or org admin can add models
    if (
        user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value
        and not _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj)
        and not await _is_user_org_admin_for_team(user_api_key_dict=user_api_key_dict, team_obj=team_obj)
    ):
        raise HTTPException(
            status_code=403,
            detail={"error": "Only proxy admin or team admin can modify team models"},
        )

    # Atomic array append with dedup at the database level so concurrent
    # BYOK model creates don't overwrite each other's team.models entries.
    # When the team currently has models=[] (unrestricted access), the
    # CASE expression inserts the 'all-proxy-models' sentinel first.
    models_to_add: Final = list(data.models)
    await prisma_client.db.execute_raw(
        'UPDATE "LiteLLM_TeamTable" '
        "SET models = ("
        "  SELECT ARRAY(SELECT DISTINCT unnest("
        "    CASE WHEN cardinality(COALESCE(models, ARRAY[]::text[])) = 0 "
        "         THEN ARRAY['all-proxy-models']::text[] "
        "         ELSE models "
        "    END || $1::text[]"
        "  ))"
        ") "
        "WHERE team_id = $2",
        models_to_add,
        data.team_id,
    )
    # Re-fetch via update (write-routed) instead of find_unique (read-routed)
    # to avoid returning stale data from a read replica. The models column
    # was already set by execute_raw above; this just retrieves the row from
    # the writer and lets Prisma bump updated_at.
    # `include` mirrors the relations the auth path consumes off the cached
    # team object so that `_refresh_cached_team` doesn't null them out.
    updated_team: Final = await _team_db(prisma_client).update(
        where={"team_id": data.team_id},
        data={"updated_at": datetime.now(timezone.utc)},
        include={"object_permission": True},
    )
    if updated_team is None:
        raise HTTPException(
            status_code=404,
            detail={"error": f"Team not found, passed team_id={data.team_id}"},
        )

    await _refresh_cached_team(
        team_row=updated_team,
        user_api_key_cache=user_api_key_cache,
        proxy_logging_obj=proxy_logging_obj,
    )

    return updated_team


@router.post(
    "/team/model/delete",
    tags=["team management"],
    dependencies=[Depends(user_api_key_auth)],
)
@management_endpoint_wrapper
async def team_model_delete(
    data: TeamModelDeleteRequest,
    http_request: Request,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    Remove models from a team's allowed model list. Only proxy admin or team admin can remove models.

    Parameters:
    - team_id: str - Required. The team to remove models from
    - models: List[str] - Required. List of models to remove from the team

    Example Request:
    ```
    curl --location 'http://0.0.0.0:4000/team/model/delete' \
    --header 'Authorization: Bearer sk-1234' \
    --header 'Content-Type: application/json' \
    --data '{
        "team_id": "team-1234",
        "models": ["gpt-4"]
    }'
    ```
    """
    from litellm.proxy.proxy_server import (
        prisma_client,
        proxy_logging_obj,
        user_api_key_cache,
    )

    if prisma_client is None:
        raise HTTPException(status_code=500, detail={"error": "No db connected"})

    # Get existing team
    team_row: Final = await _team_db(prisma_client).find_unique(where={"team_id": data.team_id})

    if team_row is None:
        raise HTTPException(
            status_code=404,
            detail={"error": f"Team not found, passed team_id={data.team_id}"},
        )

    team_obj: Final = LiteLLM_TeamTable.model_validate(team_row.model_dump())

    # Authorization check - only proxy admin, team admin, or org admin can remove models
    if (
        user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value
        and not _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj)
        and not await _is_user_org_admin_for_team(user_api_key_dict=user_api_key_dict, team_obj=team_obj)
    ):
        raise HTTPException(
            status_code=403,
            detail={"error": "Only proxy admin or team admin can modify team models"},
        )

    # Get current models list
    current_models: Final[Sequence[str]] = team_obj.models or []

    # Remove specified models
    updated_models: Final = [m for m in current_models if m not in data.models]

    # Update team. See team_model_add for the rationale on `include`.
    updated_team: Final = await _team_db(prisma_client).update(
        where={"team_id": data.team_id},
        data={"models": updated_models},
        include={"object_permission": True},
    )
    if updated_team is None:
        raise HTTPException(
            status_code=404,
            detail={"error": f"Team not found, passed team_id={data.team_id}"},
        )

    await _refresh_cached_team(
        team_row=updated_team,
        user_api_key_cache=user_api_key_cache,
        proxy_logging_obj=proxy_logging_obj,
    )

    return updated_team


@router.get(
    "/team/permissions_list",
    tags=["team management"],
    dependencies=[Depends(user_api_key_auth)],
)
@management_endpoint_wrapper
async def team_member_permissions(
    team_id: str = fastapi.Query(default=None, description="Team ID in the request parameters"),
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
) -> GetTeamMemberPermissionsResponse:
    """
    Get the team member permissions for a team
    """
    from litellm.proxy.proxy_server import (
        prisma_client,
        proxy_logging_obj,
        user_api_key_cache,
    )

    if prisma_client is None:
        raise HTTPException(status_code=500, detail={"error": "No db connected"})

    ## CHECK IF USER IS PROXY ADMIN OR TEAM ADMIN OR ORG ADMIN
    existing_team_row: Final = await get_team_object(
        team_id=team_id,
        prisma_client=prisma_client,
        user_api_key_cache=user_api_key_cache,
        parent_otel_span=None,
        proxy_logging_obj=proxy_logging_obj,
        check_cache_only=False,
        check_db_only=True,
    )

    complete_team_data: Final = LiteLLM_TeamTable.model_validate(existing_team_row.model_dump())

    # Admin Viewer follows the read-parity rule: see team permissions like
    # a Proxy Admin would. Team / org admins keep their existing scope.
    if (
        hasattr(user_api_key_dict, "user_role")
        and not _user_has_admin_view(user_api_key_dict)
        and not _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=complete_team_data)
        and not await _is_user_org_admin_for_team(user_api_key_dict=user_api_key_dict, team_obj=complete_team_data)
        and not _is_available_team(
            team_id=complete_team_data.team_id,
            user_api_key_dict=user_api_key_dict,
        )
    ):
        raise HTTPException(
            status_code=403,
            detail={
                "error": "Call not allowed. User not proxy admin OR team admin. route={}, team_id={}".format(
                    "/team/member_add",
                    complete_team_data.team_id,
                )
            },
        )

    if existing_team_row.team_member_permissions is None:
        existing_team_row.team_member_permissions = TeamMemberPermissionChecks.default_team_member_permissions()

    return GetTeamMemberPermissionsResponse(
        team_id=team_id,
        team_member_permissions=existing_team_row.team_member_permissions,
        all_available_permissions=TeamMemberPermissionChecks.get_all_available_team_member_permissions(),
    )


@router.post(
    "/team/permissions_update",
    tags=["team management"],
    dependencies=[Depends(user_api_key_auth)],
)
async def update_team_member_permissions(
    data: UpdateTeamMemberPermissionsRequest,
    http_request: Request,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
) -> LiteLLM_TeamTable:
    """
    Update the team member permissions for a team
    """
    from litellm.proxy.proxy_server import (
        prisma_client,
        proxy_logging_obj,
        user_api_key_cache,
    )

    if prisma_client is None:
        raise HTTPException(status_code=500, detail={"error": "No db connected"})

    ## CHECK IF USER IS PROXY ADMIN OR TEAM ADMIN OR ORG ADMIN
    existing_team_row: Final = await get_team_object(
        team_id=data.team_id,
        prisma_client=prisma_client,
        user_api_key_cache=user_api_key_cache,
        parent_otel_span=None,
        proxy_logging_obj=proxy_logging_obj,
        check_cache_only=False,
        check_db_only=True,
    )

    complete_team_data: Final = LiteLLM_TeamTable.model_validate(existing_team_row.model_dump())

    # Available-team self-join must NOT grant write access to team-wide
    # permission policies; only proxy/team/org admins can update them.
    if (
        hasattr(user_api_key_dict, "user_role")
        and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value
        and not _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=complete_team_data)
        and not await _is_user_org_admin_for_team(user_api_key_dict=user_api_key_dict, team_obj=complete_team_data)
    ):
        raise HTTPException(
            status_code=403,
            detail={
                "error": "Call not allowed. User not proxy admin OR team admin. route={}, team_id={}".format(
                    "/team/permissions_update",
                    complete_team_data.team_id,
                )
            },
        )
    # Update the team member permissions
    updated_team: Final = await _team_db(prisma_client).update(
        where={"team_id": data.team_id},
        data={"team_member_permissions": data.team_member_permissions},
    )
    if updated_team is None:
        raise HTTPException(
            status_code=404,
            detail={"error": f"Team not found, passed team_id={data.team_id}"},
        )

    return updated_team  # pyright: ignore[reportReturnType]  # prisma row, coerced by this route's response_model


@router.post(
    "/team/permissions_bulk_update",
    tags=["team management"],
    dependencies=[Depends(user_api_key_auth)],
    response_model=BulkUpdateTeamMemberPermissionsResponse,
)
@management_endpoint_wrapper
async def bulk_update_team_member_permissions(
    data: BulkUpdateTeamMemberPermissionsRequest,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    Append permissions to existing teams.

    Either pass team_ids to target specific teams, or set
    apply_to_all_teams=True to update every team. For each team,
    the provided permissions are merged with the team's existing
    permissions (duplicates are skipped).
    """
    from litellm.proxy.proxy_server import prisma_client

    if prisma_client is None:
        raise HTTPException(status_code=500, detail={"error": "No db connected"})

    if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value:
        raise HTTPException(
            status_code=403,
            detail={"error": "Only proxy admins can bulk-update team permissions"},
        )

    if not data.permissions:
        return {
            "message": "No permissions provided",
            "teams_updated": 0,
        }

    if not data.apply_to_all_teams and not data.team_ids:
        raise HTTPException(
            status_code=400,
            detail={"error": "Must provide team_ids or set apply_to_all_teams=true"},
        )

    if data.apply_to_all_teams and data.team_ids:
        raise HTTPException(
            status_code=400,
            detail={"error": "Cannot set both apply_to_all_teams=true and team_ids"},
        )

    permissions_to_add: Final = set(data.permissions)

    if data.team_ids:
        teams_updated = await _append_permissions_to_specific_teams(prisma_client, data.team_ids, permissions_to_add)
    else:
        teams_updated = await _append_permissions_to_all_teams(prisma_client, permissions_to_add)

    return {
        "message": "Team permissions updated successfully",
        "teams_updated": teams_updated,
        "permissions_appended": data.permissions,
    }


async def _compute_and_batch_updates(
    prisma_client, teams: "Sequence[prisma_models.LiteLLM_TeamTable]", permissions_to_add: set
) -> int:
    """Compute merged permissions and batch-write updates. Returns count of teams updated."""
    updates: Final = []
    for team in teams:
        existing = set(team.team_member_permissions or [])
        if permissions_to_add <= existing:
            continue
        merged = sorted(existing | permissions_to_add)  # normalise to alphabetical order
        updates.append((team.team_id, merged))

    if updates:
        batcher: Final = prisma_client.db.batch_()
        for team_id, merged_perms in updates:
            batcher.litellm_teamtable.update(
                where={"team_id": team_id},
                data={"team_member_permissions": merged_perms},
            )
        await batcher.commit()

    return len(updates)


async def _append_permissions_to_specific_teams(
    prisma_client: PrismaClient, team_ids: list[str], permissions_to_add: set
) -> int:
    """Fetch specific teams by ID and append permissions."""
    teams: Final = await _team_db(prisma_client).find_many(
        where={"team_id": {"in": team_ids}},
    )

    found_ids: Final = {team.team_id for team in teams}
    missing_ids: Final = set(team_ids) - found_ids
    if missing_ids:
        raise HTTPException(
            status_code=404,
            detail={"error": f"Team(s) not found: {sorted(missing_ids)}"},
        )

    return await _compute_and_batch_updates(prisma_client, teams, permissions_to_add)


async def _append_permissions_to_all_teams(prisma_client: PrismaClient, permissions_to_add: set) -> int:
    """Paginated read + batched write across all teams."""
    teams_updated = 0
    cursor = None
    BATCH_SIZE: Final = 500

    while True:
        find_args: _TeamFindManyArgs = {
            "take": BATCH_SIZE,
            "order": {"team_id": "asc"},
        }
        if cursor is not None:
            find_args["cursor"] = {"team_id": cursor}
            find_args["skip"] = 1

        teams = await _team_db(prisma_client).find_many(**find_args)

        if not teams:
            break

        teams_updated += await _compute_and_batch_updates(prisma_client, teams, permissions_to_add)

        cursor = teams[-1].team_id

        if len(teams) < BATCH_SIZE:
            break

    return teams_updated


def _daily_activity_error(*, status_code: int, message: str) -> HTTPException:
    """Single construction site for the `{"error": ...}` detail shape the
    /team/daily/activity endpoints have always returned."""
    return HTTPException(status_code=status_code, detail={"error": message})  # mutable-ok: FastAPI JSON detail


class _TeamDailyActivityScope(NamedTuple):
    team_ids: list[str] | None  # mutable-ok: downstream daily-activity signatures take str | list unions
    exclude_team_ids: list[str] | None  # mutable-ok: downstream daily-activity signatures take str | list unions
    team_alias_metadata: dict[str, dict[str, object]]  # mutable-ok: entity_metadata_field shape
    api_key_filter: str | list[str] | None  # mutable-ok: downstream daily-activity signatures take str | list unions


async def _resolve_team_daily_activity_scope(
    *,
    team_ids: str | None,
    exclude_team_ids: str | None,
    api_key: str | None,
    user_api_key_dict: UserAPIKeyAuth,
    prisma_client: PrismaClient,
    user_api_key_cache: UserApiKeyCache,
    proxy_logging_obj: ProxyLogging,
) -> _TeamDailyActivityScope:
    """Resolve which teams the caller may see and whether results must be
    narrowed to their own API keys. Shared by the paginated and aggregated
    /team/daily/activity endpoints so both enforce identical permissions."""
    # Convert comma-separated tags string to list if provided
    team_ids_list = team_ids.split(",") if team_ids else None
    exclude_team_ids_list: list[str] | None = None

    if exclude_team_ids:
        exclude_team_ids_list = exclude_team_ids.split(",") if exclude_team_ids else None

    if not _user_has_admin_view(user_api_key_dict):
        user_info: Final = await get_user_object(
            user_id=user_api_key_dict.user_id,
            prisma_client=prisma_client,
            user_id_upsert=False,
            user_api_key_cache=user_api_key_cache,
            parent_otel_span=user_api_key_dict.parent_otel_span,
            proxy_logging_obj=proxy_logging_obj,
            check_db_only=True,
        )
        if user_info is None:
            raise _daily_activity_error(status_code=404, message=f"User= {user_api_key_dict.user_id} not found")

        if team_ids_list is None:
            team_ids_list = user_info.teams
        else:
            # check if all team_ids are in user_info.teams
            for team_id in team_ids_list:
                if team_id not in user_info.teams:
                    raise _daily_activity_error(
                        status_code=404,
                        message=f"User does not belong to Team= {team_id}. Call `/user/info` to see user's teams",
                    )

    ## Fetch team aliases and check team admin status
    where_condition: Final[_TeamIdInFilter] = {}
    if team_ids_list:
        where_condition["team_id"] = {"in": list(team_ids_list)}
    team_aliases: Final = await _team_db(prisma_client).find_many(where=where_condition)
    team_alias_metadata: Final = {t.team_id: {"team_alias": _as_object(t.team_alias)} for t in team_aliases}

    # Check if user is team admin or has /team/daily/activity permission
    # If not, filter by user's API keys.
    #
    # Earlier this loop used `any-team admin -> set has_full_team_view=True
    # for the entire request`, so an admin of one team that requested
    # data for several teams would see API-key-level breakdowns for all
    # of them. Require full view on EVERY requested team — if the caller
    # only has admin/permission for a strict subset, fall back to
    # filtering the entire response by their own API keys (they can re-
    # request the admin-only teams separately to get the wider view).
    user_api_keys: list[str] | None = None
    if not _user_has_admin_view(user_api_key_dict) and team_ids_list and team_aliases:
        has_full_team_view = True
        for team_alias in team_aliases:
            team_obj = LiteLLM_TeamTable.model_validate(team_alias.model_dump())
            is_admin = _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj)
            has_perm = _team_member_has_permission(
                user_api_key_dict=user_api_key_dict,
                team_obj=team_obj,
                permission="/team/daily/activity",
            )
            if not (is_admin or has_perm):
                has_full_team_view = False
                break

        # If user does not have full team view, filter by their API keys
        if not has_full_team_view:
            # Get all API keys for this user
            user_keys: Final = await _tokens_db(prisma_client).find_many(where={"user_id": user_api_key_dict.user_id})
            user_api_keys = [key.token for key in user_keys if key.token]
            # If user has no API keys, return empty result
            if not user_api_keys:
                user_api_keys = [""]  # Use empty string to ensure no matches

    # If api_key parameter is provided, use it; otherwise use user_api_keys if set
    final_api_key_filter: str | list[str] | None = api_key
    if final_api_key_filter is None and user_api_keys is not None:
        final_api_key_filter = user_api_keys

    return _TeamDailyActivityScope(
        team_ids=team_ids_list,
        exclude_team_ids=exclude_team_ids_list,
        team_alias_metadata=team_alias_metadata,
        api_key_filter=final_api_key_filter,
    )


@router.get(
    "/team/daily/activity",
    response_model=SpendAnalyticsPaginatedResponse,
    tags=["team management"],
)
async def get_team_daily_activity(
    user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
    team_ids: str | None = None,
    start_date: str | None = None,
    end_date: str | None = None,
    model: str | None = None,
    api_key: str | None = None,
    page: int = 1,
    page_size: int = 10,
    exclude_team_ids: str | None = None,
):
    """
    Get daily activity for specific teams or all teams.

    Args:
        team_ids (Optional[str]): Comma-separated list of team IDs to filter by. If not provided, returns data for all teams.
        start_date (Optional[str]): Start date for the activity period (YYYY-MM-DD).
        end_date (Optional[str]): End date for the activity period (YYYY-MM-DD).
        model (Optional[str]): Filter by model name.
        api_key (Optional[str]): Filter by API key.
        page (int): Page number for pagination.
        page_size (int): Number of items per page.
        exclude_team_ids (Optional[str]): Comma-separated list of team IDs to exclude.
    Returns:
        SpendAnalyticsPaginatedResponse: Paginated response containing daily activity data.
    """
    from litellm.proxy.proxy_server import (
        prisma_client,
        proxy_logging_obj,
        user_api_key_cache,
    )

    if prisma_client is None:
        raise _daily_activity_error(status_code=500, message=CommonProxyErrors.db_not_connected_error.value)

    scope: Final = await _resolve_team_daily_activity_scope(
        team_ids=team_ids,
        exclude_team_ids=exclude_team_ids,
        api_key=api_key,
        user_api_key_dict=user_api_key_dict,
        prisma_client=prisma_client,
        user_api_key_cache=user_api_key_cache,
        proxy_logging_obj=proxy_logging_obj,
    )

    return await get_daily_activity(
        prisma_client=prisma_client,
        table_name="litellm_dailyteamspend",
        entity_id_field="team_id",
        entity_id=scope.team_ids,
        entity_metadata_field=scope.team_alias_metadata,
        exclude_entity_ids=scope.exclude_team_ids,
        start_date=start_date,
        end_date=end_date,
        model=model,
        api_key=scope.api_key_filter,
        page=page,
        page_size=page_size,
    )


_MAX_AGGREGATED_RANGE_DAYS: Final = 400


def _aggregated_date_range_error(start_date: str | None, end_date: str | None) -> str | None:
    """The aggregated endpoint has no pagination to bound its work, so malformed
    dates and ranges wider than the UI ever requests are rejected before querying."""
    if start_date is None or end_date is None:
        return "Please provide start_date and end_date"
    try:
        parsed_start: Final = datetime.strptime(start_date, "%Y-%m-%d").replace(tzinfo=timezone.utc)
        parsed_end: Final = datetime.strptime(end_date, "%Y-%m-%d").replace(tzinfo=timezone.utc)
    except ValueError:
        return "start_date and end_date must be valid YYYY-MM-DD dates"
    if parsed_end < parsed_start:
        return "end_date must be on or after start_date"
    if (parsed_end - parsed_start).days > _MAX_AGGREGATED_RANGE_DAYS:
        return f"Date range must be at most {_MAX_AGGREGATED_RANGE_DAYS} days"
    return None


@router.get(
    "/team/daily/activity/aggregated",
    response_model=SpendAnalyticsPaginatedResponse,
    tags=["team management"],
)
async def get_team_daily_activity_aggregated(
    user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
    team_ids: str | None = None,
    start_date: str | None = None,
    end_date: str | None = None,
    model: str | None = None,
    api_key: str | None = None,
    exclude_team_ids: str | None = None,
    timezone: int | None = None,
):
    """
    Aggregated daily activity for teams without pagination, including per-team breakdown.

    One SQL GROUPING SETS pass returns every day in the range regardless of row
    volume, so callers never reassemble pages. Same response shape as the
    paginated endpoint with page metadata pinned to a single page.

    Args:
        team_ids (Optional[str]): Comma-separated list of team IDs to filter by. If not provided, returns data for all teams.
        start_date (Optional[str]): Start date for the activity period (YYYY-MM-DD).
        end_date (Optional[str]): End date for the activity period (YYYY-MM-DD).
        model (Optional[str]): Filter by model name.
        api_key (Optional[str]): Filter by API key.
        exclude_team_ids (Optional[str]): Comma-separated list of team IDs to exclude.
        timezone (Optional[int]): Timezone offset in minutes from UTC, matching JavaScript's Date.getTimezoneOffset() convention.
    Returns:
        SpendAnalyticsPaginatedResponse: Response containing all daily activity data for the range.
    """
    from litellm.proxy.proxy_server import (
        prisma_client,
        proxy_logging_obj,
        user_api_key_cache,
    )

    if prisma_client is None:
        raise _daily_activity_error(status_code=500, message=CommonProxyErrors.db_not_connected_error.value)

    range_error: Final = _aggregated_date_range_error(start_date, end_date)
    if range_error is not None:
        raise _daily_activity_error(status_code=400, message=range_error)

    scope: Final = await _resolve_team_daily_activity_scope(
        team_ids=team_ids,
        exclude_team_ids=exclude_team_ids,
        api_key=api_key,
        user_api_key_dict=user_api_key_dict,
        prisma_client=prisma_client,
        user_api_key_cache=user_api_key_cache,
        proxy_logging_obj=proxy_logging_obj,
    )

    return await get_daily_activity_aggregated(
        prisma_client=prisma_client,
        table_name="litellm_dailyteamspend",
        entity_id_field="team_id",
        entity_id=scope.team_ids,
        entity_metadata_field=scope.team_alias_metadata,
        start_date=start_date,
        end_date=end_date,
        model=model,
        api_key=scope.api_key_filter,
        exclude_entity_ids=scope.exclude_team_ids,
        timezone_offset_minutes=timezone,
        include_entity_breakdown=True,
    )
