Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/client/credentials.py: 0%
71 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1from collections.abc import Mapping
2from typing import Any, Final
4import requests
6from .exceptions import UnauthorizedError
9class CredentialsManagementClient:
10 def __init__(self, base_url: str, api_key: str | None = None, timeout: int = 30):
11 """
12 Initialize the CredentialsManagementClient.
14 Args:
15 base_url (str): The base URL of the LiteLLM proxy server (e.g., "http://localhost:8000")
16 api_key (Optional[str]): API key for authentication. If provided, it will be sent as a Bearer token.
17 timeout (int): Request timeout in seconds (default: 30)
18 """
19 self._base_url = base_url.rstrip("/") # Remove trailing slash if present
20 self._api_key = api_key
21 self._timeout = timeout
23 def _get_headers(self) -> dict[str, str]:
24 """
25 Get the headers for API requests, including authorization if api_key is set.
27 Returns:
28 Dict[str, str]: Headers to use for API requests
29 """
30 headers: Final = {"Content-Type": "application/json"}
31 if self._api_key:
32 headers["Authorization"] = f"Bearer {self._api_key}"
33 return headers
35 def list(
36 self,
37 return_request: bool = False,
38 ) -> dict[str, Any] | requests.Request:
39 """
40 List all credentials.
42 Args:
43 return_request (bool): If True, returns the prepared request object instead of executing it
45 Returns:
46 Union[Dict[str, Any], requests.Request]: Either the response from the server or
47 a prepared request object if return_request is True
49 Raises:
50 UnauthorizedError: If the request fails with a 401 status code
51 requests.exceptions.RequestException: If the request fails with any other error
52 """
53 url: Final = f"{self._base_url}/credentials"
55 request: Final = requests.Request("GET", url, headers=self._get_headers())
57 if return_request:
58 return request
60 session: Final = requests.Session()
61 try:
62 response: Final = session.send(request.prepare(), timeout=self._timeout)
63 response.raise_for_status()
64 return response.json()
65 except requests.exceptions.HTTPError as e:
66 if e.response.status_code == 401:
67 raise UnauthorizedError(e)
68 raise
70 def create(
71 self,
72 credential_name: str,
73 credential_info: Mapping[str, object],
74 credential_values: Mapping[str, object],
75 return_request: bool = False,
76 ) -> dict[str, Any] | requests.Request:
77 """
78 Create a new credential.
80 Args:
81 credential_name (str): Name of the credential
82 credential_info (Dict[str, Any]): Additional information about the credential
83 credential_values (Dict[str, Any]): Values for the credential
84 return_request (bool): If True, returns the prepared request object instead of executing it
86 Returns:
87 Union[Dict[str, Any], requests.Request]: Either the response from the server or
88 a prepared request object if return_request is True
90 Raises:
91 UnauthorizedError: If the request fails with a 401 status code
92 requests.exceptions.RequestException: If the request fails with any other error
93 """
94 url: Final = f"{self._base_url}/credentials"
96 data: Final = {
97 "credential_name": credential_name,
98 "credential_info": credential_info,
99 "credential_values": credential_values,
100 }
102 request: Final = requests.Request("POST", url, headers=self._get_headers(), json=data)
104 if return_request:
105 return request
107 session: Final = requests.Session()
108 try:
109 response: Final = session.send(request.prepare(), timeout=self._timeout)
110 response.raise_for_status()
111 return response.json()
112 except requests.exceptions.HTTPError as e:
113 if e.response.status_code == 401:
114 raise UnauthorizedError(e)
115 raise
117 def delete(
118 self,
119 credential_name: str,
120 return_request: bool = False,
121 ) -> dict[str, Any] | requests.Request:
122 """
123 Delete a credential by name.
125 Args:
126 credential_name (str): Name of the credential to delete
127 return_request (bool): If True, returns the prepared request object instead of executing it
129 Returns:
130 Union[Dict[str, Any], requests.Request]: Either the response from the server or
131 a prepared request object if return_request is True
133 Raises:
134 UnauthorizedError: If the request fails with a 401 status code
135 requests.exceptions.RequestException: If the request fails with any other error
136 """
137 url: Final = f"{self._base_url}/credentials/{credential_name}"
139 request: Final = requests.Request("DELETE", url, headers=self._get_headers())
141 if return_request:
142 return request
144 session: Final = requests.Session()
145 try:
146 response: Final = session.send(request.prepare(), timeout=self._timeout)
147 response.raise_for_status()
148 return response.json()
149 except requests.exceptions.HTTPError as e:
150 if e.response.status_code == 401:
151 raise UnauthorizedError(e)
152 raise
154 def get(
155 self,
156 credential_name: str,
157 return_request: bool = False,
158 ) -> dict[str, Any] | requests.Request:
159 """
160 Get a credential by name.
162 Args:
163 credential_name (str): Name of the credential to retrieve
164 return_request (bool): If True, returns the prepared request object instead of executing it
166 Returns:
167 Union[Dict[str, Any], requests.Request]: Either the response from the server or
168 a prepared request object if return_request is True
170 Raises:
171 UnauthorizedError: If the request fails with a 401 status code
172 requests.exceptions.RequestException: If the request fails with any other error
173 """
174 url: Final = f"{self._base_url}/credentials/by_name/{credential_name}"
176 request: Final = requests.Request("GET", url, headers=self._get_headers())
178 if return_request:
179 return request
181 session: Final = requests.Session()
182 try:
183 response: Final = session.send(request.prepare(), timeout=self._timeout)
184 response.raise_for_status()
185 return response.json()
186 except requests.exceptions.HTTPError as e:
187 if e.response.status_code == 401:
188 raise UnauthorizedError(e)
189 raise