"""Клиент к API облака REG.RU (Рег.облако). Покрывает всё API целиком: серверы, действия над ними, снимки, образы, тарифы, регионы, SSH-ключи, IP-адреса, приватные сети, историю операций, биллинг. from regcloud_api import RegCloud api = RegCloud() # токен из REGRU_CLOUD_TOKEN for r in api.regions(): # где можно размещаться print(r["slug"], r["name"]) for i in api.distributions(): # какие операционные системы есть print(i["slug"], i["name"]) srv = api.create_server(size="c2-m4-d60-hp", image="ubuntu-24-04-amd64", region="openstack-msk1") api.wait_active(srv["id"]) Особенности платформы, обойдённые внутри, описаны в README. """ from __future__ import annotations import json import os import pathlib import time from typing import Any, Iterable import httpx from .errors import (AuthError, InsufficientFunds, Locked, NotFound, RegCloudError, Timeout, ValidationError) from .slugs import ACTIONS, IMAGE_TYPES, REGIONS, build_size, gpu_catalog, parse_size BASE_URL = "https://api.cloudvps.reg.ru" TOKEN_ENV = "REGRU_CLOUD_TOKEN" class RegCloud: """Всё API платформы одним объектом.""" def __init__(self, token: str | None = None, base_url: str = BASE_URL, timeout: float = 60.0, retries: int = 3): self.token = token or os.environ.get(TOKEN_ENV, "") if not self.token: raise AuthError(f"нет токена, передайте его явно или положите в {TOKEN_ENV}") self.base_url = base_url.rstrip("/") self.retries = retries self._http = httpx.Client(timeout=timeout, headers={ "Authorization": f"Bearer {self.token}", "Content-Type": "application/json", }) # ---------- низкий уровень ---------- def request(self, method: str, path: str, *, version: str = "v1", **kw) -> Any: """Один запрос с разбором ошибок платформы в понятные исключения.""" url = f"{self.base_url}/{version}{path}" last: Exception | None = None for attempt in range(self.retries): try: r = self._http.request(method, url, **kw) except httpx.HTTPError as exc: last = RegCloudError(f"сеть подвела: {type(exc).__name__}: {exc}") time.sleep(2 * (attempt + 1)) continue if r.status_code < 400: if not r.content: return {} try: return r.json() except json.JSONDecodeError: return r.text body: dict = {} try: body = r.json() except json.JSONDecodeError: body = {"message": r.text[:400]} code = str(body.get("code") or "") message = str(body.get("message") or body.get("detail") or r.text[:300]) if r.status_code in (401, 403): raise AuthError(message, r.status_code, code, body) if r.status_code == 404 or code in ("NO_SUCH_REGLET", "IMAGE_NOT_FOUND"): raise NotFound(message, r.status_code, code, body) if code in ("RESOURCE_DELETE_ERROR", "RESOURCE_LOCKED", "SERVER_IS_STOPPED"): raise Locked(message, r.status_code, code, body) if "funds" in message.lower() or "баланс" in message.lower(): raise InsufficientFunds(message, r.status_code, code, body) if r.status_code == 400: raise ValidationError(message, r.status_code, code, body) last = RegCloudError(message, r.status_code, code, body) time.sleep(2 * (attempt + 1)) raise last or RegCloudError("запрос не удался") def get(self, path, **kw): return self.request("GET", path, **kw) def post(self, path, **kw): return self.request("POST", path, **kw) def put(self, path, **kw): return self.request("PUT", path, **kw) def delete(self, path, **kw): return self.request("DELETE", path, **kw) # ---------- справочники ---------- def regions(self) -> list[dict]: """Где можно размещать серверы. Отдельного метода у платформы нет, список зашит и подтверждён её же проверкой при создании сервера.""" return [{"slug": s, "name": n} for s, n in REGIONS.items()] def sizes(self, region: str = "") -> list[dict]: """Тарифы. Учтите, что платформа возвращает не все линейки, тарифы с видеокартами сюда не попадают, для них есть gpu_sizes().""" params = {"region": region} if region else None data = self.get("/sizes", params=params) return data.get("sizes", []) if isinstance(data, dict) else [] def gpu_sizes(self) -> list[dict]: """Тарифы с видеокартой. Собственные конфигурации для них платформа не принимает, только эти готовые.""" return gpu_catalog() def size_info(self, slug: str) -> dict: """Разобрать slug тарифа на понятные значения.""" return parse_size(slug) @staticmethod def make_size(cpu: int, ram: int, disk: int, line: str = "hp") -> str: """Собрать slug тарифа с проверкой допустимых сочетаний.""" return build_size(cpu, ram, disk, line) def images(self, *, type: str = "", region: str = "", group: str = "", private: bool | None = None, reglet_id: int | None = None) -> list[dict]: """Образы. Тип бывает distribution, application, snapshot, backup, restore.""" if type and type not in IMAGE_TYPES: raise ValueError(f"тип бывает {', '.join(IMAGE_TYPES)}") params = {k: v for k, v in (("type", type), ("region", region), ("group", group), ("private", private), ("reglet_id", reglet_id)) if v not in ("", None)} data = self.get("/images", params=params or None) return data.get("images", []) if isinstance(data, dict) else [] def distributions(self, region: str = "") -> list[dict]: """Только операционные системы, без готовых сборок.""" return self.images(type="distribution", region=region) def applications(self, region: str = "") -> list[dict]: """Готовые сборки: панели, докер, гитлаб и прочее.""" return self.images(type="application", region=region) def images_v2(self, region: str, *, type: str = "", page: int = 1, per_page: int = 100) -> list[dict]: """Образы через вторую версию API. Там их больше и они разложены по регионам.""" params = {"region": region, "page": page, "items_per_page": per_page} if type: params["type"] = type data = self.request("GET", "/images", version="v2", params=params) return data.get("images", data.get("data", [])) if isinstance(data, dict) else [] def prices(self) -> list[dict]: data = self.get("/prices") return data.get("prices", []) if isinstance(data, dict) else [] def random_name(self) -> str: """Платформа умеет придумывать имена серверам сама.""" data = self.get("/random_reglet_name") return data.get("name", "") if isinstance(data, dict) else str(data) # ---------- серверы ---------- def servers(self) -> list[dict]: data = self.get("/reglets") return data.get("reglets", []) if isinstance(data, dict) else [] list = servers # привычное имя тоже оставим def server(self, server_id: int) -> dict: data = self.get(f"/reglets/{server_id}") return data.get("reglet", data) def removed_servers(self) -> list[dict]: """Удалённые серверы. Полезно, когда надо понять, что и когда снесли.""" data = self.get("/removed_servers") return data.get("reglets", data.get("servers", [])) if isinstance(data, dict) else [] def create_server(self, *, size: str, image: str | int, name: str = "", region: str = "", ssh_keys: Iterable[str] | None = None, backups: bool = False, floating_ip: bool = False, isp_license: str = "", promocode: str = "") -> dict: """Создать сервер. В image можно передать как slug образа, так и номер снимка.""" if region and region not in REGIONS: raise ValueError(f"нет региона {region}, есть {', '.join(REGIONS)}") payload: dict[str, Any] = {"size": size, "image": image} if name: payload["name"] = name if region: payload["region_slug"] = region if ssh_keys: payload["ssh_keys"] = list(ssh_keys) if backups: payload["backups"] = True if floating_ip: payload["floating_ip"] = True if isp_license: payload["isp_license_size"] = isp_license if promocode: payload["promocode"] = promocode data = self.post("/reglets", json=payload) return data.get("reglet", data) create = create_server def rename_server(self, server_id: int, name: str) -> dict: return self.put(f"/reglets/{server_id}", json={"name": name}) def destroy_server(self, server_id: int, *, patient: bool = True, timeout: float = 900) -> bool: """Удалить сервер. Сразу после снимка сервер сидит в locked и удаляться отказывается, отпускает примерно через полминуты. По умолчанию повторяем попытки, потому что забытый сервер продолжает тарифицироваться.""" deadline = time.time() + timeout while True: try: self.delete(f"/reglets/{server_id}") return True except NotFound: return True # его уже нет, это успех except Locked: if not patient or time.time() > deadline: raise time.sleep(20) destroy = destroy_server # ---------- действия над сервером ---------- def action(self, server_id: int, type: str, **params) -> dict: if type not in ACTIONS: raise ValueError(f"действие бывает {', '.join(ACTIONS)}") data = self.post(f"/reglets/{server_id}/actions", json={"type": type, **params}) return data.get("action", data) def start(self, server_id: int) -> dict: return self.action(server_id, "start") def stop(self, server_id: int) -> dict: return self.action(server_id, "stop") def reboot(self, server_id: int) -> dict: return self.action(server_id, "reboot") def reset_password(self, server_id: int) -> dict: return self.action(server_id, "password_reset") def rebuild(self, server_id: int, image: str | int) -> dict: """Переустановить систему на сервере.""" return self.action(server_id, "rebuild", image=image) def resize(self, server_id: int, size: str) -> dict: """Сменить тариф.""" return self.action(server_id, "resize", size=size) def clone(self, server_id: int, name: str = "") -> dict: return self.action(server_id, "clone", **({"name": name} if name else {})) def restore(self, server_id: int, image: str | int) -> dict: """Откатить сервер на снимок или бэкап.""" return self.action(server_id, "restore", image=image) def vnc_link(self, server_id: int) -> dict: """Ссылка на консоль, когда по SSH уже не зайти.""" return self.action(server_id, "generate_vnc_link") def set_backups(self, server_id: int, on: bool) -> dict: return self.action(server_id, "enable_backups" if on else "disable_backups") def snapshot(self, server_id: int, name: str, offline: bool = True) -> dict: """Снять образ. Сервер при этом должен работать, платформа остановит его сама, если передан offline. С уже остановленного она снимать отказывается.""" return self.action(server_id, "snapshot", name=name, offline=1 if offline else 0) # ---------- ожидание ---------- def wait_active(self, server_id: int, timeout: float = 900, poll: float = 5) -> dict: """Дождаться, пока сервер поднимется и получит адрес.""" deadline = time.time() + timeout while time.time() < deadline: info = self.server(server_id) if info.get("status") == "active" and self.ip_of(info): return info time.sleep(poll) raise Timeout(f"сервер {server_id} не поднялся за {timeout} секунд") def wait_status(self, server_id: int, status: str, timeout: float = 600, poll: float = 5) -> dict: deadline = time.time() + timeout while time.time() < deadline: info = self.server(server_id) if info.get("status") == status: return info time.sleep(poll) raise Timeout(f"сервер {server_id} не пришёл в состояние {status}") def wait_action(self, action_id: int, timeout: float = 900, poll: float = 5) -> dict: deadline = time.time() + timeout while time.time() < deadline: data = self.get(f"/actions/{action_id}") act = data.get("action", data) if act.get("status") in ("completed", "errored"): return act time.sleep(poll) raise Timeout(f"действие {action_id} не завершилось за {timeout} секунд") def wait_snapshot(self, name: str, timeout: float = 1800, poll: float = 10) -> dict: """Дождаться появления снимка с таким именем.""" deadline = time.time() + timeout while time.time() < deadline: for s in self.snapshots(): if s.get("name") == name: return s time.sleep(poll) raise Timeout(f"снимок {name} не появился за {timeout} секунд") # ---------- снимки ---------- def snapshots(self, region: str = "") -> list[dict]: params = {"region": region} if region else None data = self.get("/snapshots", params=params) return data.get("snapshots", []) if isinstance(data, dict) else [] def delete_snapshot(self, image_id: int) -> bool: self.delete(f"/snapshots/{image_id}") return True def servers_from_snapshot(self, snapshot_id: int) -> list[dict]: """Какие серверы подняты из этого снимка.""" data = self.get(f"/reglets_for_snapshot/{snapshot_id}") return data.get("reglets", []) if isinstance(data, dict) else [] # ---------- ключи доступа ---------- def ssh_keys(self) -> list[dict]: data = self.get("/account/keys") return data.get("ssh_keys", []) if isinstance(data, dict) else [] def add_ssh_key(self, name: str, public_key: str | pathlib.Path) -> dict: """Публичный ключ можно передать текстом или путём к файлу.""" key = public_key path = pathlib.Path(str(public_key)) if path.exists(): key = path.read_text().strip() data = self.post("/account/keys", json={"name": name, "public_key": str(key).strip()}) return data.get("ssh_key", data) def rename_ssh_key(self, key_id: int, name: str) -> dict: return self.put(f"/account/keys/{key_id}", json={"name": name}) def delete_ssh_key(self, key_id: int) -> bool: self.delete(f"/account/keys/{key_id}") return True def key_fingerprint(self, name: str) -> str: """Отпечаток ключа по его имени, он нужен при создании сервера.""" for k in self.ssh_keys(): if k.get("name") == name: return k.get("fingerprint", "") raise NotFound(f"нет ключа с именем {name}") # ---------- адреса ---------- def ips(self, server_id: int | None = None) -> list[dict]: params = {"reglet_id": server_id} if server_id else None data = self.get("/ips", params=params) return data.get("ips", data) if isinstance(data, dict) else data def add_ip(self, server_id: int, ipv4: int = 1, ipv6: int = 0) -> dict: return self.post("/ips", json={"reglet_id": server_id, "ipv4_count": ipv4, "ipv6_count": ipv6}) def delete_ip(self, ip: str) -> bool: self.delete(f"/ips/{ip}") return True def set_ptr(self, ip: str, ptr: str) -> dict: """Обратная запись для адреса, нужна для почты.""" return self.put(f"/ips/{ip}", json={"ip": ip, "ptr": ptr}) # ---------- приватные сети ---------- # # Приватная сеть связывает серверы напрямую, минуя интернет. Трафик внутри неё # не тарифицируется и не выходит наружу, поэтому машину-исполнителя можно держать # вообще без публичного адреса, а управляющий сервер разговаривает с ней по локали. # # Что выяснено на живом API и чего нет в документации. # Номер сети лежит в поле resource_id, а не id. # Список сетей приходит голым массивом, без обёртки. # При создании принимается только имя. Регион, участники и resource_id # объявлены только для чтения, платформа проставляет их сама. # Подключение сервера к сети это асинхронное действие, в ответ приходит его карточка. # Пока действие не доиграет, сервер заблокирован, и отключить его или удалить сеть # не выйдет, поэтому обе операции тут настойчивые. def vpcs(self) -> list[dict]: """Приватные сети. Номер сети лежит в поле resource_id.""" data = self.get("/vpcs") if isinstance(data, list): return data return data.get("vpcs", []) if isinstance(data, dict) else [] def vpc(self, vpc_id: int) -> dict: data = self.get(f"/vpcs/{vpc_id}") return data.get("vpc", data) if isinstance(data, dict) else data def create_vpc(self, name: str) -> dict: """Создать приватную сеть. Кроме имени платформа ничего не принимает. Регион она выставляет сама, участники добавляются потом через vpc_attach().""" data = self.post("/vpcs", json={"name": name}) return data.get("vpc", data) if isinstance(data, dict) else data def rename_vpc(self, vpc_id: int, name: str) -> dict: return self.put(f"/vpcs/{vpc_id}", json={"name": name}) def delete_vpc(self, vpc_id: int, patient: bool = True, timeout: float = 300) -> bool: """Удалить сеть. Если её участники заняты, ждём и повторяем.""" deadline = time.time() + timeout while True: try: self.delete(f"/vpcs/{vpc_id}") return True except NotFound: return True except (Locked, ValidationError): if not patient or time.time() > deadline: raise time.sleep(15) def vpc_members(self, vpc_id: int) -> list[int]: """Номера серверов в сети.""" data = self.get(f"/vpcs/{vpc_id}/members") if isinstance(data, list): return data return data.get("members", []) if isinstance(data, dict) else [] def vpc_attach(self, vpc_id: int, server_id: int) -> dict: """Подключить сервер к сети. Действие асинхронное, сервер на время блокируется.""" data = self.post(f"/vpcs/{vpc_id}/members", json={"resource_id": server_id}) return data.get("action", data) if isinstance(data, dict) else data def vpc_detach(self, vpc_id: int, server_id: int, patient: bool = True, timeout: float = 300) -> bool: """Отключить сервер от сети. Если он занят другой операцией, ждём.""" deadline = time.time() + timeout while True: try: self.delete(f"/vpcs/{vpc_id}/members/{server_id}") return True except NotFound: return True except (Locked, ValidationError): if not patient or time.time() > deadline: raise time.sleep(15) def vpc_wait_member(self, vpc_id: int, server_id: int, timeout: float = 300) -> bool: """Дождаться, пока сервер действительно окажется в сети.""" deadline = time.time() + timeout while time.time() < deadline: if server_id in self.vpc_members(vpc_id): return True time.sleep(5) raise Timeout(f"сервер {server_id} не появился в сети {vpc_id}") @staticmethod def vpc_id_of(vpc: dict) -> int: """Номер сети. Платформа кладёт его в resource_id, а не в id.""" return int(vpc.get("resource_id") or vpc.get("id") or 0) # ---------- kubernetes ---------- def kubeconfig(self, cluster_id: int) -> str: data = self.get(f"/k8s_clusters/{cluster_id}/get_kubeconfig") return data.get("kubeconfig", data) if isinstance(data, dict) else str(data) # ---------- деньги и история ---------- def balance(self) -> dict: data = self.get("/balance_data") return data.get("balance_data", data) def money_left(self) -> float: return float(self.balance().get("balance") or 0) def burn_rate(self) -> dict: """Сколько сейчас капает в час и в сутки по всем работающим услугам.""" b = self.balance() per_hour = sum(float(i.get("price") or 0) for row in b.get("detalization", []) for i in row.get("linked", [])) return {"в час": round(per_hour, 4), "в сутки": round(per_hour * 24, 2), "в месяц": round(per_hour * 24 * 30, 2), "баланс": float(b.get("balance") or 0), "дней осталось": b.get("days_left")} def billing_history(self) -> list[dict]: data = self.get("/billing_history") return data.get("billing_history", data) if isinstance(data, dict) else data def history(self, limit: int = 50, offset: int = 0) -> list[dict]: """История операций над ресурсами.""" data = self.get("/history", params={"limit": limit, "offset": offset}) return data.get("history", data) if isinstance(data, dict) else data # ---------- разное ---------- def settings(self, key: str) -> Any: return self.get(f"/account/settings/{key}") def set_setting(self, key: str, value: Any) -> Any: return self.put(f"/account/settings/{key}", json={"value": value}) def feedback(self, message: str, email: str = "") -> dict: return self.post("/account/feedback", json={"message": message, "user_email": email}) @staticmethod def ip_of(server: dict) -> str: """Адрес сервера. Лежит в поле ip, а networks у платформы всегда пустое.""" if server.get("ip"): return server["ip"] nets = (server.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 close(self) -> None: self._http.close() def __enter__(self): return self def __exit__(self, *a): self.close()