70 lines
2.8 KiB
Python
70 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
|
|
class UEXError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class UEXClient:
|
|
def __init__(self, base_url: str, secret_key: str | None = None, bearer_token: str | None = None) -> None:
|
|
self.base_url = base_url.rstrip("/")
|
|
self.secret_key = secret_key
|
|
self.bearer_token = bearer_token
|
|
|
|
def _headers(self, authenticated: bool = False) -> dict[str, str]:
|
|
headers = {"Accept": "application/json"}
|
|
if authenticated:
|
|
if not self.secret_key and not self.bearer_token:
|
|
raise UEXError("UEX_SECRET_KEY or UEX_BEARER_TOKEN is required for this action.")
|
|
if self.secret_key:
|
|
headers["secret-key"] = self.secret_key
|
|
if self.bearer_token:
|
|
headers["Authorization"] = f"Bearer {self.bearer_token}"
|
|
return headers
|
|
|
|
async def get(self, path: str, params: dict[str, Any] | None = None, authenticated: bool = False) -> dict[str, Any]:
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
|
response = await client.get(
|
|
f"{self.base_url}/{path.strip('/')}/",
|
|
params={k: v for k, v in (params or {}).items() if v is not None},
|
|
headers=self._headers(authenticated),
|
|
)
|
|
return self._handle_response(response)
|
|
|
|
async def get_user(self, username: str | None = None, authenticated: bool = False) -> dict[str, Any]:
|
|
body = await self.get("user", {"username": username}, authenticated=authenticated)
|
|
data = body.get("data")
|
|
if isinstance(data, list):
|
|
data = data[0] if data else None
|
|
return {"status": body.get("status"), "user": data}
|
|
|
|
async def get_user_notifications(self) -> dict[str, Any]:
|
|
body = await self.get("user_notifications", authenticated=True)
|
|
data = body.get("data") or []
|
|
if isinstance(data, dict):
|
|
data = [data]
|
|
return {"status": body.get("status"), "notifications": data}
|
|
|
|
async def post(self, path: str, payload: dict[str, Any], authenticated: bool = True) -> dict[str, Any]:
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
|
response = await client.post(
|
|
f"{self.base_url}/{path.strip('/')}/",
|
|
json=payload,
|
|
headers=self._headers(authenticated),
|
|
)
|
|
return self._handle_response(response)
|
|
|
|
@staticmethod
|
|
def _handle_response(response: httpx.Response) -> dict[str, Any]:
|
|
try:
|
|
body = response.json()
|
|
except ValueError as exc:
|
|
raise UEXError(f"UEX returned non-JSON response: HTTP {response.status_code}") from exc
|
|
if response.status_code >= 400:
|
|
raise UEXError(f"UEX HTTP {response.status_code}: {body}")
|
|
return body
|