requests 封装即可覆盖几乎全部使用场景。将下面的类复制到你的项目中即可。
安装
pip install requests
客户端
aria_compute.py
import os
from typing import Optional
import requests
class AriaCompute:
def __init__(self, api_key: str, base_url: str = "https://ariacompute.cn/api"):
self.base_url = base_url.rstrip("/")
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Accept": "application/json",
})
def _get(self, path: str, **kwargs):
r = self.session.get(f"{self.base_url}{path}", **kwargs)
r.raise_for_status()
return r.json()
def _post(self, path: str, json=None, **kwargs):
r = self.session.post(f"{self.base_url}{path}", json=json, **kwargs)
r.raise_for_status()
return r.json()
# Models
def list_models(self):
return self._get("/models")
def download_model(self, slug: str, quant: str, sdk: str, dest: str):
url = f"{self.base_url}/models/{slug}/download"
params = {"quant": quant, "sdk": sdk}
with self.session.get(url, params=params, stream=True, allow_redirects=True) as r:
r.raise_for_status()
with open(dest, "wb") as f:
for chunk in r.iter_content(chunk_size=1 << 20):
f.write(chunk)
# Billing
def wallet(self):
return self._get("/billing/wallet")
def create_payment(self, provider: str, amount: float, currency: str):
return self._post("/billing/payments", json={
"provider": provider, "amount": amount, "currency": currency,
})
用法
client = AriaCompute(api_key=os.environ["ARIA_API_KEY"])
for model in client.list_models()["models"]:
print(model["slug"])
client.download_model(
slug="gemma-4-e2b-it",
quant="int4",
sdk="v1.0",
dest="./gemma-4-e2b-it_q4.zip",
)
print(client.wallet())
若使用国际站,请传入
base_url="https://ariacompute.com/api"。账户、钱包与密钥均按区域隔离:在 .cn 签发的密钥无法在 .com 使用。重试与超时
后端返回标准 HTTP 状态码。用requests.adapters.HTTPAdapter 配合 urllib3.util.Retry 封装调用,以重试瞬时 5xx 与 429 响应,并始终设置 timeout:
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
retry = Retry(total=5, backoff_factor=0.5, status_forcelist=[429, 500, 502, 503, 504])
client.session.mount("https://", HTTPAdapter(max_retries=retry))
client.session.request = lambda method, url, **kw: requests.Session.request(
client.session, method, url, timeout=kw.pop("timeout", 30), **kw
)