mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-29 06:33:28 +00:00
130699ca23
* refactor(typing): narrow Any sources in proxy, guardrails, bedrock and responses code Turns untyped json.loads, response.json(), header lookups and forwarded kwargs into object or precise types at their source, so downstream code type-checks without reportAny / reportExplicitAny. No runtime behavior changes. * refactor(typing): narrow Any sources in caching, MCP, providers and proxy request code Decoded JSON, cache reads and header lookups now enter as object or precise types, prisma tables used by the shadow-eval logger get typed Protocol views, and a few SDK parameters move from dict[str, Any] to Mapping[str, object]. No runtime behavior changes. * refactor(typing): narrow Any sources in proxy server, router, logging and shared utils * refactor(typing): narrow Any sources in provider transformations, integrations and proxy endpoints * refactor(typing): narrow Any sources across providers, proxy and router Deepens the Any cleanup on the hottest files: typed helpers in core_helpers for attribute and item access, identity TypeAdapter(object) wraps at untyped boundaries, explicit return annotations on shared provider and proxy helpers, and Protocols in place of Any parameters. No intended behavior change.
74 lines
3.2 KiB
Python
74 lines
3.2 KiB
Python
from collections.abc import Iterable, Mapping
|
|
from typing import Any, Final
|
|
|
|
import requests
|
|
|
|
from .exceptions import NotFoundError, UnauthorizedError
|
|
|
|
|
|
class UsersManagementClient:
|
|
def __init__(self, base_url: str, api_key: str | None = None, timeout: int = 30):
|
|
self.base_url = base_url.rstrip("/")
|
|
self.api_key = api_key
|
|
self.timeout = timeout
|
|
|
|
def _get_headers(self) -> dict[str, str]:
|
|
headers: Final = {"Content-Type": "application/json"}
|
|
if self.api_key:
|
|
headers["Authorization"] = f"Bearer {self.api_key}"
|
|
return headers
|
|
|
|
def list_users(
|
|
self, params: Mapping[str, str | bytes | int | float | Iterable[str | bytes | int | float] | None] | None = None
|
|
) -> list[dict[str, Any]]:
|
|
"""List users (GET /user/list)"""
|
|
url: Final = f"{self.base_url}/user/list"
|
|
response: Final = requests.get(url, headers=self._get_headers(), params=params, timeout=self.timeout)
|
|
if response.status_code == 401:
|
|
raise UnauthorizedError(response.text)
|
|
response.raise_for_status()
|
|
return response.json().get("users", response.json())
|
|
|
|
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 {}
|
|
response: Final = requests.get(url, headers=self._get_headers(), params=params, timeout=self.timeout)
|
|
if response.status_code == 401:
|
|
raise UnauthorizedError(response.text)
|
|
if response.status_code == 404:
|
|
raise NotFoundError(response.text)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
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 {}
|
|
response: Final = requests.get(url, headers=self._get_headers(), params=params, timeout=self.timeout)
|
|
if response.status_code == 401:
|
|
raise UnauthorizedError(response.text)
|
|
if response.status_code == 404:
|
|
raise NotFoundError(response.text)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
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 response.json()
|
|
|
|
def delete_user(self, user_ids: list[str]) -> dict[str, object]:
|
|
"""Delete users (POST /user/delete)"""
|
|
url: Final = f"{self.base_url}/user/delete"
|
|
response: Final = requests.post(
|
|
url, headers=self._get_headers(), json={"user_ids": user_ids}, timeout=self.timeout
|
|
)
|
|
if response.status_code == 401:
|
|
raise UnauthorizedError(response.text)
|
|
response.raise_for_status()
|
|
return response.json()
|