viktor
a106ba5b6b
GPU это лишь один из тарифов, а клиент общего назначения. Примеры переписаны на обычные серверы, добавлен пример с масштабированием под нагрузку. Claude-Session: https://claude.ai/code/session_01Loz5stgQZH6ofBdMqTxiJu
228 строки
10 KiB
Python
228 строки
10 KiB
Python
"""Клиент к API облака REG.RU (Рег.облако).
|
|
|
|
API устроено как у DigitalOcean, сервер там называется reglet.
|
|
Токен берётся в личном кабинете, в разделе настроек.
|
|
|
|
from regcloud_api import RegCloud
|
|
|
|
api = RegCloud() # токен из REGRU_CLOUD_TOKEN
|
|
for s in api.list():
|
|
print(s["id"], s["name"], s["status"])
|
|
|
|
Зачем обёртка, если API простое. Затем, что у него есть неочевидные места,
|
|
собранные на живых серверах и обойдённые тут.
|
|
|
|
Адрес сервера лежит в поле ip, а не в networks, оно всегда пустое.
|
|
Снимок снимается только с работающего сервера, останавливает его платформа сама по offline.
|
|
Сразу после снимка сервер сидит в locked и удаляться отказывается примерно полминуты,
|
|
поэтому удаление настойчивое, иначе забытый сервер продолжит тарифицироваться.
|
|
Метод sizes возвращает не все линейки тарифов, полный список смотрите в панели управления.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import time
|
|
|
|
import httpx
|
|
|
|
BASE = "https://api.cloudvps.reg.ru/v1"
|
|
STATE_FILE = pathlib.Path(os.environ.get("REGCLOUD_STATE_DIR", ".")) / "regcloud_state.json"
|
|
|
|
|
|
class RegCloud:
|
|
def __init__(self, token: str | None = None, timeout: float = 60.0):
|
|
self.token = token or os.environ.get("REGRU_CLOUD_TOKEN", "")
|
|
if not self.token:
|
|
raise RuntimeError("нет токена, положи REGRU_CLOUD_TOKEN в окружение")
|
|
self.http = httpx.Client(
|
|
base_url=BASE, timeout=timeout,
|
|
headers={"Authorization": f"Bearer {self.token}", "Content-Type": "application/json"},
|
|
)
|
|
|
|
# ---------- справочники ----------
|
|
|
|
def sizes(self) -> list[dict]:
|
|
r = self.http.get("/sizes")
|
|
r.raise_for_status()
|
|
return r.json().get("sizes", [])
|
|
|
|
def gpu_sizes(self) -> list[dict]:
|
|
"""Отбор тарифов с видеокартой, если они вам нужны.
|
|
Учтите, что sizes() возвращает не все линейки, полный список в панели управления."""
|
|
out = []
|
|
for s in self.sizes():
|
|
blob = json.dumps(s, ensure_ascii=False).lower()
|
|
if "gpu" in blob or "nvidia" in blob or "a4000" in blob or "a5000" in blob or "a100" in blob:
|
|
out.append(s)
|
|
return out
|
|
|
|
def images(self) -> list[dict]:
|
|
r = self.http.get("/images")
|
|
r.raise_for_status()
|
|
return r.json().get("images", [])
|
|
|
|
def snapshots(self) -> list[dict]:
|
|
r = self.http.get("/snapshots")
|
|
r.raise_for_status()
|
|
return r.json().get("snapshots", [])
|
|
|
|
def ssh_keys(self) -> list[dict]:
|
|
r = self.http.get("/account/keys")
|
|
r.raise_for_status()
|
|
return r.json().get("ssh_keys", [])
|
|
|
|
def balance(self) -> dict:
|
|
r = self.http.get("/balance_data")
|
|
return r.json() if r.status_code == 200 else {"error": r.status_code, "text": r.text[:200]}
|
|
|
|
def prices(self) -> dict:
|
|
r = self.http.get("/prices")
|
|
return r.json() if r.status_code == 200 else {}
|
|
|
|
# ---------- машины ----------
|
|
|
|
def list(self) -> list[dict]:
|
|
r = self.http.get("/reglets")
|
|
r.raise_for_status()
|
|
return r.json().get("reglets", [])
|
|
|
|
def get(self, reglet_id: int) -> dict:
|
|
r = self.http.get(f"/reglets/{reglet_id}")
|
|
r.raise_for_status()
|
|
return r.json().get("reglet", {})
|
|
|
|
def create(self, size: str, image: str | int, name: str = "gpuhub-worker",
|
|
ssh_keys: list[str] | None = None, region: str = "") -> dict:
|
|
payload: dict = {"name": name, "size": size, "image": image}
|
|
if ssh_keys:
|
|
payload["ssh_keys"] = ssh_keys
|
|
if region:
|
|
payload["region_slug"] = region
|
|
r = self.http.post("/reglets", json=payload)
|
|
if r.status_code >= 400:
|
|
raise RuntimeError(f"не создалась машина, {r.status_code} {r.text[:300]}")
|
|
return r.json().get("reglet", {})
|
|
|
|
def destroy(self, reglet_id: int, patient: bool = True, timeout: float = 900) -> bool:
|
|
"""Удаление машины.
|
|
|
|
Сразу после снятия снимка машина сидит в locked и удаляться отказывается,
|
|
отвечая RESOURCE_DELETE_ERROR. Проверено на живой машине, отпускает примерно
|
|
через полминуты. Поэтому по умолчанию долбимся, пока не поддастся, иначе
|
|
забытая машина будет тикать по деньгам.
|
|
"""
|
|
deadline = time.time() + timeout
|
|
while True:
|
|
r = self.http.delete(f"/reglets/{reglet_id}")
|
|
if r.status_code < 400 or r.status_code == 404:
|
|
return True
|
|
body = r.text[:200]
|
|
if "NO_SUCH_REGLET" in body:
|
|
return True # её уже нет, это успех
|
|
if not patient or time.time() > deadline:
|
|
raise RuntimeError(f"не удалилась машина, {r.status_code} {body}")
|
|
time.sleep(20)
|
|
|
|
def action(self, reglet_id: int, action_type: str, **extra) -> dict:
|
|
r = self.http.post(f"/reglets/{reglet_id}/actions", json={"type": action_type, **extra})
|
|
if r.status_code >= 400:
|
|
raise RuntimeError(f"действие {action_type} не прошло, {r.status_code} {r.text[:200]}")
|
|
return r.json()
|
|
|
|
def snapshot(self, reglet_id: int, name: str, offline: bool = True) -> dict:
|
|
"""Снимок машины. Из него потом поднимаются готовые к работе клоны.
|
|
Снимаем с выключенной, так образ выходит целостным."""
|
|
return self.action(reglet_id, "snapshot", name=name, offline=offline)
|
|
|
|
def stop(self, reglet_id: int) -> dict:
|
|
"""Остановка. Важно понимать, что у большинства провайдеров остановленная машина
|
|
всё равно занимает диск и продолжает тарифицироваться. Для экономии её нужно удалять."""
|
|
return self.action(reglet_id, "stop")
|
|
|
|
def start(self, reglet_id: int) -> dict:
|
|
return self.action(reglet_id, "start")
|
|
|
|
def wait_action(self, action_id: int, timeout: float = 900) -> dict:
|
|
"""Ждём завершения долгого действия, например снятия снимка."""
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
r = self.http.get(f"/actions/{action_id}")
|
|
if r.status_code == 200:
|
|
data = r.json().get("action", {})
|
|
if data.get("status") in ("completed", "errored"):
|
|
return data
|
|
time.sleep(5)
|
|
raise TimeoutError(f"действие {action_id} не завершилось за {timeout} секунд")
|
|
|
|
def wait_active(self, reglet_id: int, timeout: float = 600) -> dict:
|
|
"""Ждём, пока машина поднимется и получит адрес."""
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
info = self.get(reglet_id)
|
|
if info.get("status") == "active" and self.ip_of(info):
|
|
return info
|
|
time.sleep(5)
|
|
raise TimeoutError(f"машина {reglet_id} не поднялась за {timeout} секунд")
|
|
|
|
@staticmethod
|
|
def ip_of(reglet: dict) -> str:
|
|
"""Адрес лежит прямо в поле ip, а networks у этого провайдера пустует."""
|
|
if reglet.get("ip"):
|
|
return reglet["ip"]
|
|
nets = (reglet.get("networks") or {}).get("v4") or []
|
|
for n in nets:
|
|
if n.get("type") == "public":
|
|
return n.get("ip_address", "")
|
|
return nets[0].get("ip_address", "") if nets else ""
|
|
|
|
|
|
# ---------- память о поднятых машинах ----------
|
|
|
|
def load_state() -> dict:
|
|
if STATE_FILE.exists():
|
|
try:
|
|
return json.loads(STATE_FILE.read_text())
|
|
except json.JSONDecodeError:
|
|
pass
|
|
return {"reglet_id": None, "ip": "", "started_at": 0, "idle_since": 0}
|
|
|
|
|
|
def save_state(state: dict) -> None:
|
|
STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
STATE_FILE.write_text(json.dumps(state, ensure_ascii=False, indent=1))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
api = RegCloud()
|
|
cmd = sys.argv[1] if len(sys.argv) > 1 else "sizes"
|
|
|
|
if cmd == "sizes":
|
|
for s in api.gpu_sizes():
|
|
price = s.get("price_hourly") or s.get("price_monthly") or "?"
|
|
print(f"{s.get('slug','?'):28s} {s.get('description','')[:50]:52s} {price}")
|
|
elif cmd == "all-sizes":
|
|
for s in api.sizes():
|
|
print(f"{s.get('slug','?'):28s} {s.get('description','')[:60]}")
|
|
elif cmd == "images":
|
|
for i in api.images()[:40]:
|
|
print(f"{str(i.get('id','')):10s} {i.get('slug','') or i.get('name',''):35s} {i.get('distribution','')}")
|
|
elif cmd == "snapshots":
|
|
for s in api.snapshots():
|
|
print(f"{s.get('id')} {s.get('name')} {s.get('size_gigabytes')} ГБ")
|
|
elif cmd == "list":
|
|
for r in api.list():
|
|
print(f"{r.get('id')} {r.get('name'):25s} {r.get('status'):10s} {RegCloud.ip_of(r)}")
|
|
elif cmd == "keys":
|
|
for k in api.ssh_keys():
|
|
print(f"{k.get('id')} {k.get('name')} {k.get('fingerprint')}")
|
|
elif cmd == "balance":
|
|
print(json.dumps(api.balance(), ensure_ascii=False, indent=1))
|
|
elif cmd == "prices":
|
|
data = api.prices()
|
|
print(json.dumps(data, ensure_ascii=False, indent=1)[:3000])
|
|
else:
|
|
print("sizes | all-sizes | images | snapshots | list | keys | balance | prices")
|
© 2026 StackWorks