Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/hooks/user_management_event_hooks.py: 68%
60 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""
2Hooks that are triggered when a litellm user event occurs
3"""
5import asyncio
6from datetime import datetime, timezone
7from typing import Final
9import litellm
10from litellm._logging import verbose_proxy_logger
11from litellm._uuid import uuid
12from litellm.proxy._types import (
13 AUDIT_ACTIONS,
14 CommonProxyErrors,
15 LiteLLM_AuditLogs,
16 Litellm_EntityType,
17 LitellmTableNames,
18 NewUserRequest,
19 NewUserResponse,
20 UserAPIKeyAuth,
21 WebhookEvent,
22)
23from litellm.proxy.management_helpers.audit_logs import (
24 create_audit_log_for_update,
25 is_audit_logging_enabled,
26)
27from litellm.repositories.user_repository import UserRepository
30class UserManagementEventHooks:
31 @staticmethod
32 async def async_user_created_hook(
33 data: NewUserRequest,
34 response: NewUserResponse,
35 user_api_key_dict: UserAPIKeyAuth,
36 ):
37 """
38 This hook is called when a new user is created on litellm
40 Handles:
41 - Creating an audit log for the user creation
42 - Sending a user invitation email to the user
43 """
44 from litellm.proxy.proxy_server import litellm_proxy_admin_name, prisma_client
46 #########################################################
47 ########## Send User Invitation Email ################
48 #########################################################
49 await UserManagementEventHooks.async_send_user_invitation_email(
50 data=data,
51 response=response,
52 user_api_key_dict=user_api_key_dict,
53 )
55 #########################################################
56 ########## CREATE AUDIT LOG ################
57 #########################################################
58 try:
59 if prisma_client is None: 59 ↛ 60line 59 didn't jump to line 60 because the condition on line 59 was never true
60 raise Exception(CommonProxyErrors.db_not_connected_error.value)
61 if response.user_id is None: 61 ↛ 62line 61 didn't jump to line 62 because the condition on line 61 was never true
62 raise Exception("no user_id returned for the newly created user")
63 user_row: Final = await UserRepository(prisma_client).find_by_id(response.user_id)
64 if user_row is None: 64 ↛ 65line 64 didn't jump to line 65 because the condition on line 64 was never true
65 raise Exception(f"no user row found for user_id={response.user_id}")
66 asyncio.create_task(
67 UserManagementEventHooks.create_internal_user_audit_log(
68 user_id=user_row.user_id,
69 action="created",
70 litellm_changed_by=user_api_key_dict.user_id,
71 user_api_key_dict=user_api_key_dict,
72 litellm_proxy_admin_name=litellm_proxy_admin_name,
73 before_value=None,
74 after_value=user_row.model_dump_json(exclude_none=True),
75 )
76 )
77 except Exception as e:
78 verbose_proxy_logger.warning("Unable to create audit log for user on `/user/new` - %s", e)
80 @staticmethod
81 async def async_send_user_invitation_email(
82 data: NewUserRequest,
83 response: NewUserResponse,
84 user_api_key_dict: UserAPIKeyAuth,
85 ):
86 """
87 Send a user invitation email to the user
88 """
89 event: Final = WebhookEvent(
90 event="internal_user_created",
91 event_group=Litellm_EntityType.USER,
92 event_message="Welcome to LiteLLM Proxy",
93 token=response.token,
94 spend=response.spend or 0.0,
95 max_budget=response.max_budget,
96 user_id=response.user_id,
97 user_email=response.user_email,
98 team_id=response.team_id,
99 key_alias=response.key_alias,
100 )
102 sent_via_v2: Final = await UserManagementEventHooks._send_v2_user_invitation_emails(
103 event=event, send_invite_email=data.send_invite_email
104 )
106 #########################################################
107 ########## LEGACY V1 USER INVITATION EMAIL (FALLBACK) ####
108 #########################################################
109 if data.send_invite_email is True and not sent_via_v2:
110 await UserManagementEventHooks.send_legacy_v1_user_invitation_email(
111 data=data,
112 response=response,
113 user_api_key_dict=user_api_key_dict,
114 event=event,
115 )
117 @staticmethod
118 async def _send_v2_user_invitation_emails(event: WebhookEvent, send_invite_email: bool | None) -> bool:
119 """
120 Send the modern (V2) invitation email via any registered enterprise email logger.
122 Returns True if at least one logger delivered, so the caller only falls back to
123 the legacy email when V2 did not send (enterprise package absent, no email logger
124 configured, or every send raised).
125 """
126 if send_invite_email is not True:
127 return False
129 try:
130 from litellm_enterprise.enterprise_callbacks.send_emails.base_email import (
131 BaseEmailLogger,
132 )
133 except ImportError:
134 verbose_proxy_logger.warning(
135 "Defaulting to using Legacy Email Hooks." + CommonProxyErrors.missing_enterprise_package.value
136 )
137 return False
139 email_loggers: Final = tuple(
140 email_logger
141 for email_logger in litellm.logging_callback_manager.get_custom_loggers_for_type(
142 callback_type=BaseEmailLogger
143 )
144 if isinstance(email_logger, BaseEmailLogger)
145 )
146 if len(email_loggers) == 0: 146 ↛ 149line 146 didn't jump to line 149 because the condition on line 146 was always true
147 return False
149 send_outcomes: Final = await asyncio.gather(
150 *(email_logger.send_user_invitation_email(event=event) for email_logger in email_loggers),
151 return_exceptions=True,
152 )
153 for outcome in send_outcomes:
154 if isinstance(outcome, BaseException):
155 verbose_proxy_logger.error(
156 "Error sending v2 user invitation email for user_id=%s: %s",
157 event.user_id,
158 str(outcome),
159 )
161 return any(not isinstance(outcome, BaseException) for outcome in send_outcomes)
163 @staticmethod
164 async def send_legacy_v1_user_invitation_email(
165 data: NewUserRequest,
166 response: NewUserResponse,
167 user_api_key_dict: UserAPIKeyAuth,
168 event: WebhookEvent,
169 ):
170 """
171 Send a user invitation email to the user
172 """
173 from litellm.proxy.proxy_server import general_settings, proxy_logging_obj
175 # check if user has setup email alerting
176 if "email" not in general_settings.get("alerting", []): 176 ↛ 182line 176 didn't jump to line 182 because the condition on line 176 was always true
177 raise ValueError(
178 "Email alerting not setup on config.yaml. Please set `alerting=['email']. \nDocs: https://docs.litellm.ai/docs/proxy/email`"
179 )
181 # If user configured email alerting - send an Email letting their end-user know the key was created
182 asyncio.create_task(
183 proxy_logging_obj.slack_alerting_instance.send_key_created_or_user_invited_email(
184 webhook_event=event,
185 )
186 )
188 @staticmethod
189 async def create_internal_user_audit_log(
190 user_id: str,
191 action: AUDIT_ACTIONS,
192 litellm_changed_by: str | None,
193 user_api_key_dict: UserAPIKeyAuth,
194 litellm_proxy_admin_name: str | None,
195 before_value: str | None = None,
196 after_value: str | None = None,
197 ):
198 """
199 Create an audit log for an internal user.
201 Parameters:
202 - user_id: str - The id of the user to create the audit log for.
203 - action: AUDIT_ACTIONS - The action to create the audit log for.
204 - user_row: LiteLLM_UserTable - The user row to create the audit log for.
205 - litellm_changed_by: Optional[str] - The user id of the user who is changing the user.
206 - user_api_key_dict: UserAPIKeyAuth - The user api key dictionary.
207 - litellm_proxy_admin_name: Optional[str] - The name of the proxy admin.
208 """
209 if not is_audit_logging_enabled(): 209 ↛ 212line 209 didn't jump to line 212 because the condition on line 209 was always true
210 return
212 from litellm.proxy.management_helpers.audit_logs import (
213 get_audit_log_changed_by,
214 )
216 await create_audit_log_for_update(
217 request_data=LiteLLM_AuditLogs(
218 id=str(uuid.uuid4()),
219 updated_at=datetime.now(timezone.utc),
220 changed_by=get_audit_log_changed_by(
221 litellm_changed_by=litellm_changed_by,
222 user_api_key_dict=user_api_key_dict,
223 litellm_proxy_admin_name=litellm_proxy_admin_name,
224 ),
225 changed_by_api_key=user_api_key_dict.api_key,
226 table_name=LitellmTableNames.USER_TABLE_NAME,
227 object_id=user_id,
228 action=action,
229 updated_values=after_value,
230 before_value=before_value,
231 )
232 )