Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/client/users.py: 0%

1from collections.abc import Mapping 

2from typing import Any, Final 

3 

4import requests 

5 

6from .exceptions import NotFoundError, UnauthorizedError 

7 

8 

9class UsersManagementClient: 

10 def __init__(self, base_url: str, api_key: str | None = None, timeout: int = 30): 

11 self.base_url = base_url.rstrip("/") 

12 self.api_key = api_key 

13 self.timeout = timeout 

14 

15 def _get_headers(self) -> dict[str, str]: 

16 headers: Final = {"Content-Type": "application/json"} 

17 if self.api_key: 

18 headers["Authorization"] = f"Bearer {self.api_key}" 

19 return headers 

20 

21 def list_users(self, params: dict[str, Any] | None = None) -> list[dict[str, Any]]: 

22 """List users (GET /user/list)""" 

23 url: Final = f"{self.base_url}/user/list" 

24 response: Final = requests.get(url, headers=self._get_headers(), params=params, timeout=self.timeout) 

25 if response.status_code == 401: 

26 raise UnauthorizedError(response.text) 

27 response.raise_for_status() 

28 return response.json().get("users", response.json()) 

29 

30 def get_user(self, user_id: str | None = None) -> dict[str, Any]: 

31 """Get user info (GET /user/info)""" 

32 url: Final = f"{self.base_url}/user/info" 

33 params: Final = {"user_id": user_id} if user_id else {} 

34 response: Final = requests.get(url, headers=self._get_headers(), params=params, timeout=self.timeout) 

35 if response.status_code == 401: 

36 raise UnauthorizedError(response.text) 

37 if response.status_code == 404: 

38 raise NotFoundError(response.text) 

39 response.raise_for_status() 

40 return response.json() 

41 

42 def get_user_v2(self, user_id: str | None = None) -> dict[str, Any]: 

43 """Get user info v2 - lightweight, returns only user object (GET /v2/user/info)""" 

44 url: Final = f"{self.base_url}/v2/user/info" 

45 params: Final = {"user_id": user_id} if user_id else {} 

46 response: Final = requests.get(url, headers=self._get_headers(), params=params, timeout=self.timeout) 

47 if response.status_code == 401: 

48 raise UnauthorizedError(response.text) 

49 if response.status_code == 404: 

50 raise NotFoundError(response.text) 

51 response.raise_for_status() 

52 return response.json() 

53 

54 def create_user(self, user_data: Mapping[str, object]) -> dict[str, Any]: 

55 """Create a new user (POST /user/new)""" 

56 url: Final = f"{self.base_url}/user/new" 

57 response: Final = requests.post(url, headers=self._get_headers(), json=user_data, timeout=self.timeout) 

58 if response.status_code == 401: 

59 raise UnauthorizedError(response.text) 

60 response.raise_for_status() 

61 return response.json() 

62 

63 def delete_user(self, user_ids: list[str]) -> dict[str, Any]: 

64 """Delete users (POST /user/delete)""" 

65 url: Final = f"{self.base_url}/user/delete" 

66 response: Final = requests.post( 

67 url, headers=self._get_headers(), json={"user_ids": user_ids}, timeout=self.timeout 

68 ) 

69 if response.status_code == 401: 

70 raise UnauthorizedError(response.text) 

71 response.raise_for_status() 

72 return response.json()