"""Клиент к 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, оно всегда пустое. Снимок нельзя снять с остановленной машины, API сам её останавливает по offline. Сразу после снимка машина сидит в locked и удаляться отказывается примерно полминуты, поэтому удаление тут настойчивое, иначе забытая машина будет тикать по деньгам. Тарифы с видеокартами метод sizes не показывает вовсе, хотя создание с ними работает, их список лежит в docs_slugs.md. """ 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]: """Тарифы с видеокартой. У них в описании или в названии есть упоминание gpu.""" 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")