60 lines
2.4 KiB
Python
60 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import asdict, is_dataclass
|
|
from typing import Any
|
|
from urllib.error import HTTPError
|
|
from urllib.request import Request, urlopen
|
|
|
|
from .errors import APIError, BusinessError
|
|
|
|
|
|
def csv_join(items: list[str]) -> str:
|
|
return ",".join(item.strip() for item in items if item.strip())
|
|
|
|
|
|
class Client:
|
|
def __init__(self, base_url: str, token: str, timeout: float = 15.0) -> None:
|
|
self.base_url = base_url.rstrip("/")
|
|
self.token = token
|
|
self.timeout = timeout if timeout > 0 else 15.0
|
|
self.account_type = "stock"
|
|
|
|
def set_account_type(self, account_type: str) -> "Client":
|
|
if account_type.strip():
|
|
self.account_type = account_type
|
|
return self
|
|
|
|
def _request(self, method: str, path: str, body: Any = None) -> Any:
|
|
data = None
|
|
headers = {"X-Token": self.token, "Accept": "application/json"}
|
|
if method != "GET":
|
|
if body is None: body = {}
|
|
if is_dataclass(body): body = asdict(body)
|
|
data = json.dumps(body, ensure_ascii=False).encode()
|
|
headers["Content-Type"] = "application/json"
|
|
request = Request(self.base_url + path, data=data, headers=headers, method=method)
|
|
try:
|
|
with urlopen(request, timeout=self.timeout) as response:
|
|
raw = response.read()
|
|
except HTTPError as exc:
|
|
raw = exc.read()
|
|
try: message = json.loads(raw).get("error", raw.decode(errors="replace"))
|
|
except (ValueError, AttributeError): message = raw.decode(errors="replace").strip()
|
|
raise APIError(exc.code, str(message)) from exc
|
|
if not raw: return None
|
|
try: return json.loads(raw)
|
|
except ValueError as exc: raise ValueError(f"invalid JSON from {path}: {raw[:512]!r}") from exc
|
|
|
|
def _get(self, path: str) -> Any: return self._request("GET", path)
|
|
def _post(self, path: str, body: Any = None) -> Any: return self._request("POST", path, body)
|
|
|
|
def _get_field(self, path: str, key: str) -> Any:
|
|
return self._get(path).get(key)
|
|
|
|
def _post_field(self, path: str, body: Any, key: str) -> Any:
|
|
result = self._post(path, body)
|
|
if isinstance(result, dict) and result.get("error"):
|
|
raise BusinessError(result["error"])
|
|
return result.get(key, result) if key and isinstance(result, dict) else result
|