fix(proxy): fall back to a team admin's inline email when the user record has none

This commit is contained in:
Rameshwaran Mohan 2026-09-15 05:39:47 +05:30
parent 121e3ba7e2
commit c0305a568d
2 changed files with 25 additions and 3 deletions

View file

@ -3,6 +3,7 @@ Functions for sending Email Alerts
"""
import os
from types import MappingProxyType
from typing import Final
from litellm._logging import verbose_logger, verbose_proxy_logger
@ -68,9 +69,8 @@ async def get_all_team_member_emails(team_id: str | None = None) -> list:
async def get_team_admin_emails(team: LiteLLM_TeamTable, prisma_client: object) -> tuple[str, ...]:
"""Emails of the team's admins: by user id, inline email, or the legacy admins list, deduped"""
"""Emails of the team's admins: the user record's email, else the inline one, plus the legacy admins list, deduped"""
admins: Final = tuple(member for member in team.members_with_roles if member.role == "admin")
inline_emails: Final = tuple(member.user_email for member in admins if member.user_email and not member.user_id)
admin_ids: Final = frozenset(
(*(member.user_id for member in admins if member.user_id), *(user_id for user_id in team.admins if user_id))
)
@ -81,7 +81,11 @@ async def get_team_admin_emails(team: LiteLLM_TeamTable, prisma_client: object)
if admin_ids
else ()
)
return tuple(dict.fromkeys((*inline_emails, *(row.user_email for row in rows if row.user_email))))
record_emails: Final = MappingProxyType({row.user_id: row.user_email for row in rows if row.user_email})
inline_emails: Final = tuple(
member.user_email for member in admins if member.user_email and member.user_id not in record_emails
)
return tuple(dict.fromkeys((*inline_emails, *record_emails.values())))
async def send_team_budget_alert(webhook_event: WebhookEvent) -> bool:

View file

@ -101,3 +101,21 @@ async def test_should_email_an_email_only_admin_without_querying_the_database():
)
assert await get_team_admin_emails(team, prisma) == ("mail-only@example.com",)
assert prisma.db.litellm_usertable.queries == []
@pytest.mark.asyncio
async def test_should_fall_back_to_the_inline_email_when_the_user_record_has_none():
team: Final = LiteLLM_TeamTable(
team_id="t1",
members_with_roles=[Member(user_id="u-no-email", user_email="inline@example.com", role="admin")],
)
assert await get_team_admin_emails(team, _prisma(USERS)) == ("inline@example.com",)
@pytest.mark.asyncio
async def test_should_prefer_the_user_record_email_over_a_differing_inline_one():
team: Final = LiteLLM_TeamTable(
team_id="t1",
members_with_roles=[Member(user_id="u-admin", user_email="stale@example.com", role="admin")],
)
assert await get_team_admin_emails(team, _prisma(USERS)) == ("admin@example.com",)