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

1"""Teams management client for LiteLLM proxy.""" 

2 

3import builtins 

4from typing import Any, Final 

5 

6import requests 

7 

8from .exceptions import UnauthorizedError 

9 

10 

11class TeamsManagementClient: 

12 """Client for managing teams in LiteLLM proxy.""" 

13 

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

15 """ 

16 Initialize the TeamsManagementClient. 

17 

18 Args: 

19 base_url (str): The base URL of the LiteLLM proxy server (e.g., "http://localhost:4000") 

20 api_key (Optional[str]): API key for authentication. If provided, it will be sent as a Bearer token. 

21 timeout (int): Request timeout in seconds (default: 30) 

22 """ 

23 self._base_url = base_url.rstrip("/") # Remove trailing slash if present 

24 self._api_key = api_key 

25 self._timeout = timeout 

26 

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

28 """ 

29 Get the headers for API requests, including authorization if api_key is set. 

30 

31 Returns: 

32 Dict[str, str]: Headers to use for API requests 

33 """ 

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

35 if self._api_key: 

36 headers["Authorization"] = f"Bearer {self._api_key}" 

37 return headers 

38 

39 def list( 

40 self, 

41 user_id: str | None = None, 

42 organization_id: str | None = None, 

43 ) -> builtins.list[dict[str, Any]]: 

44 """ 

45 List teams that the user belongs to. 

46 

47 Args: 

48 user_id (Optional[str]): Only return teams which this user belongs to 

49 organization_id (Optional[str]): Only return teams which belong to this organization 

50 

51 Returns: 

52 List[Dict[str, Any]]: List of team objects 

53 

54 Raises: 

55 requests.exceptions.HTTPError: If the request fails 

56 UnauthorizedError: If authentication fails 

57 """ 

58 url: Final = f"{self._base_url}/team/list" 

59 params: Final = {} 

60 if user_id: 

61 params["user_id"] = user_id 

62 if organization_id: 

63 params["organization_id"] = organization_id 

64 

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

66 

67 if response.status_code == 401: 

68 raise UnauthorizedError("Authentication failed. Check your API key.") 

69 

70 response.raise_for_status() 

71 return response.json() 

72 

73 def list_v2( 

74 self, 

75 user_id: str | None = None, 

76 organization_id: str | None = None, 

77 team_id: str | None = None, 

78 team_alias: str | None = None, 

79 page: int = 1, 

80 page_size: int = 10, 

81 sort_by: str | None = None, 

82 sort_order: str = "asc", 

83 ) -> dict[str, Any]: 

84 """ 

85 Get a paginated list of teams with filtering and sorting options. 

86 

87 Args: 

88 user_id (Optional[str]): Only return teams which this user belongs to 

89 organization_id (Optional[str]): Only return teams which belong to this organization 

90 team_id (Optional[str]): Filter teams by exact team_id match 

91 team_alias (Optional[str]): Filter teams by partial team_alias match 

92 page (int): Page number for pagination 

93 page_size (int): Number of teams per page 

94 sort_by (Optional[str]): Column to sort by (e.g. 'team_id', 'team_alias', 'created_at') 

95 sort_order (str): Sort order ('asc' or 'desc') 

96 

97 Returns: 

98 Dict[str, Any]: Paginated response containing teams and pagination info 

99 

100 Raises: 

101 requests.exceptions.HTTPError: If the request fails 

102 UnauthorizedError: If authentication fails 

103 """ 

104 url: Final = f"{self._base_url}/v2/team/list" 

105 params: Final[dict[str, str | int]] = { 

106 "page": page, 

107 "page_size": page_size, 

108 "sort_order": sort_order, 

109 } 

110 

111 if user_id: 

112 params["user_id"] = user_id 

113 if organization_id: 

114 params["organization_id"] = organization_id 

115 if team_id: 

116 params["team_id"] = team_id 

117 if team_alias: 

118 params["team_alias"] = team_alias 

119 if sort_by: 

120 params["sort_by"] = sort_by 

121 

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

123 

124 if response.status_code == 401: 

125 raise UnauthorizedError("Authentication failed. Check your API key.") 

126 

127 response.raise_for_status() 

128 return response.json() 

129 

130 def get_available(self) -> builtins.list[dict[str, Any]]: 

131 """ 

132 Get list of available teams that the user can join. 

133 

134 Returns: 

135 List[Dict[str, Any]]: List of available team objects 

136 

137 Raises: 

138 requests.exceptions.HTTPError: If the request fails 

139 UnauthorizedError: If authentication fails 

140 """ 

141 url: Final = f"{self._base_url}/team/available" 

142 

143 response: Final = requests.get(url, headers=self._get_headers(), timeout=self._timeout) 

144 

145 if response.status_code == 401: 

146 raise UnauthorizedError("Authentication failed. Check your API key.") 

147 

148 response.raise_for_status() 

149 return response.json()