scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-11 08:38:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 280

fix(types): validate Python client user responses and type delete_user as the count it returns

Browse source
This commit is contained in:
mateo-berri 2026-10-28 06:12:59 -07:00
parent 8d7fe995b1
commit 659af8d73d
1 changed files with 7 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
litellm/proxy/client/users.py
View file

@ -7,6 +7,7 @@ from pydantic import ConfigDict, TypeAdapter
from .exceptions import NotFoundError, UnauthorizedError
_JSON_VALUE: Final = TypeAdapter(object, config=ConfigDict(strict=True))
_JSON_OBJECT: Final = TypeAdapter(dict[str, object], config=ConfigDict(strict=True))
class UsersManagementClient:
@ -32,7 +33,7 @@ class UsersManagementClient:
response.raise_for_status()
return response.json().get("users", response.json())
def get_user(self, user_id: str | None = None) -> object:
def get_user(self, user_id: str | None = None) -> dict[str, object]:
"""Get user info (GET /user/info)"""
url: Final = f"{self.base_url}/user/info"
params: Final = {"user_id": user_id} if user_id else {}
@ -42,9 +43,9 @@ class UsersManagementClient:
if response.status_code == 404:
raise NotFoundError(response.text)
response.raise_for_status()
return _JSON_VALUE.validate_python(response.json())
return _JSON_OBJECT.validate_python(response.json())
def get_user_v2(self, user_id: str | None = None) -> object:
def get_user_v2(self, user_id: str | None = None) -> dict[str, object]:
"""Get user info v2 - lightweight, returns only user object (GET /v2/user/info)"""
url: Final = f"{self.base_url}/v2/user/info"
params: Final = {"user_id": user_id} if user_id else {}
@ -54,16 +55,16 @@ class UsersManagementClient:
if response.status_code == 404:
raise NotFoundError(response.text)
response.raise_for_status()
return _JSON_VALUE.validate_python(response.json())
return _JSON_OBJECT.validate_python(response.json())
def create_user(self, user_data: Mapping[str, object]) -> object:
def create_user(self, user_data: Mapping[str, object]) -> dict[str, object]:
"""Create a new user (POST /user/new)"""
url: Final = f"{self.base_url}/user/new"
response: Final = requests.post(url, headers=self._get_headers(), json=user_data, timeout=self.timeout)
if response.status_code == 401:
raise UnauthorizedError(response.text)
response.raise_for_status()
return _JSON_VALUE.validate_python(response.json())
return _JSON_OBJECT.validate_python(response.json())
def delete_user(self, user_ids: list[str]) -> object:
"""Delete users (POST /user/delete)"""