viktor/regcloud-api
1
0
Форкнуть 0
Код Задачи Запросы на слияние Действия Пакеты Проекты Релизы Вики Активность

Полное покрытие API

Просмотр исходного кода 
Клиент дописан до всех двадцати семи эндпоинтов платформы.

Добавлено:
- регионы, образы с разбивкой на системы и готовые сборки, вторая версия API образов
- все действия над сервером: старт, стоп, перезагрузка, переустановка, смена тарифа,
  клонирование, откат, консоль, бэкапы
- снимки целиком, включая ожидание готовности и поиск поднятых из них серверов
- ключи доступа, IP-адреса с обратными записями, приватные сети, kubeconfig
- история операций, биллинг, расход в час и в сутки
- сборка и проверка slug тарифа по правилам платформы
- ошибки разложены по смыслу, чтобы ловить прицельно

Claude-Session: https://claude.ai/code/session_01Loz5stgQZH6ofBdMqTxiJu
Этот коммит содержится в:
viktor
2026-08-23 18:30:07 +09:00
родитель a106ba5b6b
Коммит 7e798bb68f
8 изменённых файлов: 831 добавлений и 259 удалений
Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы
README.md
+114 -63
Просмотреть файл
@@ -2,8 +2,9 @@
Клиент к API облака REG.RU (Рег.облако), точка входа `https://api.cloudvps.reg.ru`.
Создание и удаление серверов, снимки, SSH-ключи, тарифы, образы, баланс и цены.
Само API устроено как у DigitalOcean, сервер там называется reglet.
Полное покрытие API: серверы и все действия над ними, снимки, образы, тарифы, регионы,
SSH-ключи, IP-адреса и обратные записи, приватные сети, kubeconfig кластеров,
история операций и биллинг.
## Установка
@@ -12,104 +13,154 @@ pip install git+https://git.stackworks.ru/viktor/regcloud-api.git
```
Токен берётся в личном кабинете, в разделе настроек, и кладётся в `REGRU_CLOUD_TOKEN`.
Либо передаётся в конструктор явно.
## Быстрый старт
## Осмотреться
```python
from regcloud_api import RegCloud
api = RegCloud()
for s in api.list():
print(s["id"], s["name"], s["status"], RegCloud.ip_of(s))
for r in api.regions(): # где можно размещаться
print(r["slug"], r["name"])
print(api.balance())
for i in api.distributions(): # операционные системы
print(i["slug"], i["distribution"])
for i in api.applications(): # готовые сборки: панели, докер, гитлаб
print(i["slug"])
for s in api.sizes(): # тарифы
print(s["slug"], s["vcpus"], "ядер", s["price"], "р/час")
```
Поднять сервер и дождаться готовности.
Во второй версии API образов заметно больше и они разложены по регионам.
```python
srv = api.create(
size="c2-m4-d60-hp", # 2 ядра, 4 ГБ, 60 ГБ, линейка «производительный»
for i in api.images_v2("openstack-spb1", type="distribution"):
print(i["id"], i["name"])
```
## Тарифы
Slug собирается по шаблону `c<ядра>-m<память>-d<диск>-<линейка>`. Клиент умеет
собирать его сам и заодно проверять, что такое сочетание вообще существует.
```python
from regcloud_api import build_size, parse_size
RegCloud.make_size(4, 8, 120, "hp") # 'c4-m8-d120-hp'
build_size(4, 5, 120, "hp") # ValueError, при 4 ядрах памяти 5 ГБ не бывает
parse_size("c8-m32-d128-g1-a4000-gpu")
# {'gpu': 'A4000 16 ГБ', 'vcpus': 8, 'memory_gb': 32, 'disk_gb': 128, 'price_hour': 37.8}
```
Линейки: `std` стандартный, `stdp` стандартный плюс, `hp` производительный,
`hps` производительный ФЗ-152, `ddc` выделенный CPU, `high` высокочастотный.
## Жизненный цикл сервера
```python
key = api.add_ssh_key("deploy", "~/.ssh/id_ed25519.pub") # можно путём к файлу
srv = api.create_server(
size="c2-m4-d60-hp",
image="ubuntu-24-04-amd64",
name="web-01",
name=api.random_name(), # платформа умеет придумывать имена
region="openstack-msk1",
ssh_keys=["d7:3a:b4:..."], # отпечаток ключа из api.ssh_keys()
ssh_keys=[key["fingerprint"]],
)
info = api.wait_active(srv["id"])
print("адрес:", RegCloud.ip_of(info))
```
Снять образ и поднять из него копию.
Действия над сервером.
```python
api.snapshot(srv["id"], "web-01-base", offline=True) # сервер остановит сам
copy = api.create(size="c2-m4-d60-hp", image=7887485, region="openstack-msk1")
api.stop(id); api.start(id); api.reboot(id)
api.resize(id, "c4-m8-d120-hp") # сменить тариф
api.rebuild(id, "debian-12-amd64") # переустановить систему
api.reset_password(id)
api.vnc_link(id) # консоль, когда по ssh уже не зайти
api.set_backups(id, True)
api.clone(id, "копия")
```
Удалить.
Снимки.
```python
api.destroy(srv["id"])
api.snapshot(id, "base", offline=True) # платформа сама остановит сервер
snap = api.wait_snapshot("base")
api.create_server(size="c2-m4-d60-hp", image=snap["id"], region="openstack-msk1")
api.servers_from_snapshot(snap["id"]) # кто уже поднят из этого снимка
api.delete_snapshot(snap["id"])
```
## Методы
Удаление.
| Метод | Что делает |
|---|---|
| `list()`, `get(id)` | серверы на аккаунте |
| `create(size, image, name, ssh_keys, region)` | создать сервер, из образа или из снимка |
| `destroy(id)` | удалить |
| `stop(id)`, `start(id)` | выключить, включить |
| `action(id, type, **kw)` | любое действие: reboot, rebuild, resize, clone, restore и прочие |
| `snapshot(id, name, offline)` | снять образ |
| `wait_active(id)`, `wait_action(id)` | дождаться готовности сервера или завершения действия |
| `sizes()`, `gpu_sizes()` | тарифы |
| `images()`, `snapshots()` | образы и снимки |
| `ssh_keys()` | ключи доступа |
| `balance()`, `prices()` | деньги |
| `RegCloud.ip_of(server)` | вытащить адрес из карточки сервера |
```python
api.destroy_server(id) # дождётся снятия блокировки
```
## Особенности API, которые тут уже обойдены
## Адреса и сети
Это не претензии к платформе, просто места, где поведение отличается от ожидаемого.
Собрано на живых серверах, чтобы вы на них не наступали.
```python
api.ips(server_id)
api.add_ip(server_id, ipv4=1)
api.set_ptr("1.2.3.4", "mail.example.com")
api.delete_ip("1.2.3.4")
**Адрес сервера лежит в поле `ip`.** Поле `networks`, которое по документации должно
его содержать, на практике всегда пустое. `ip_of()` смотрит в оба места.
vpc = api.create_vpc("backoffice", region="openstack-msk1")
api.vpc_attach(vpc["id"], server_id)
api.vpc_members(vpc["id"])
api.vpc_detach(vpc["id"], server_id)
```
**Снимок снимается только с работающего сервера.** У остановленного API отвечает
`SERVER_IS_STOPPED`. Останавливать его для снимка должна сама платформа,
для этого есть параметр `offline`.
## Деньги
**Сразу после снимка сервер сидит в `locked`** и на удаление отвечает
`RESOURCE_DELETE_ERROR`. Отпускает примерно через полминуты. Поэтому `destroy()`
по умолчанию повторяет попытки, пока не получится, и считает `NO_SUCH_REGLET`
успехом. Забытый сервер продолжает тарифицироваться, так что молча сдаваться тут нельзя.
```python
api.money_left() # сколько на счету
api.burn_rate() # сколько капает в час, в сутки и в месяц
api.billing_history()
api.history(limit=50) # что и когда делали с ресурсами
api.prices()
```
**`sizes()` показывает не все тарифы.** Часть линеек метод не возвращает, хотя создание
с ними работает нормально. Полный список смотрите в панели управления, в разделе
настроек, пункт со slug'ами для API и Terraform. Наша выборка лежит в [docs_slugs.md](docs_slugs.md).
## Ошибки
Все ошибки платформы разложены по смыслу, чтобы ловить прицельно.
```python
from regcloud_api import NotFound, Locked, ValidationError, InsufficientFunds
try:
api.destroy_server(123)
except NotFound:
pass # его уже нет, и хорошо
except Locked:
pass # занят другой операцией
```
## Особенности платформы, обойдённые внутри
Собрано на живых серверах, чтобы вы на это не наступали.
**Адрес сервера лежит в поле `ip`.** Поле `networks` из документации всегда пустое.
`RegCloud.ip_of()` смотрит в оба места.
**Снимок снимается только с работающего сервера.** У остановленного платформа отвечает
`SERVER_IS_STOPPED`, останавливать его должна она сама по параметру `offline`.
**После снимка сервер сидит в `locked`** и на удаление отвечает `RESOURCE_DELETE_ERROR`.
Отпускает примерно через полминуты, поэтому `destroy_server()` повторяет попытки.
Забытый сервер продолжает тарифицироваться, молча сдаваться тут нельзя.
**`sizes()` возвращает не все линейки.** Тарифы с видеокартами в выдачу не попадают,
хотя создание с ними работает. Их список лежит в `gpu_sizes()` и в [docs_slugs.md](docs_slugs.md).
**Тарификация идёт от часа.** Двадцать минут работы списываются как полный час.
Короткие запуски невыгодны, задачи лучше копить и выполнять пачкой.
## Составление slug тарифа
`c<ядра>-m<память>-d<диск>-<линейка>`, например `c4-m8-d120-hp`.
Линейки: `std` стандартный, `stdp` стандартный плюс, `hp` производительный,
`hps` производительный ФЗ-152, `ddc` выделенный CPU, `high` высокочастотный.
Допустимые сочетания ядер, памяти и диска смотрите в [docs_slugs.md](docs_slugs.md).
## Регионы
`openstack-msk1`, `openstack-msk2`, `openstack-msk3`, `openstack-spb1`,
`openstack-sam1`, `openstack-fz1` и архивный `msk1`.
Не все образы и тарифы доступны в каждом регионе.
## Рядом
examples/golden_image.py
+25
Просмотреть файл
@@ -0,0 +1,25 @@
"""Эталонный образ: настроить машину один раз, снять снимок, дальше клонировать.
Так разворачивание клиента занимает минуты вместо часа настройки.
"""
from regcloud_api import RegCloud
api = RegCloud()
REGION = "openstack-msk1"
SIZE = "c2-m4-d60-hp"
srv = api.create_server(size=SIZE, image="ubuntu-24-04-amd64", name="golden", region=REGION)
info = api.wait_active(srv["id"])
print("настраивайте машину по адресу", RegCloud.ip_of(info))
input("нажмите Enter, когда закончите...")
api.snapshot(srv["id"], "golden-v1", offline=True)
snap = api.wait_snapshot("golden-v1")
print("снимок готов:", snap["id"], snap.get("size_gigabytes"), "ГБ")
api.destroy_server(srv["id"])
print("эталон удалён, снимок остался")
for n in range(1, 4):
copy = api.create_server(size=SIZE, image=snap["id"], name=f"client-{n:02d}", region=REGION)
print("поднял копию", copy["id"])
examples/inventory.py
+26
Просмотреть файл
@@ -0,0 +1,26 @@
"""Осмотреться на аккаунте: что где размещается, что запущено, куда уходят деньги."""
from regcloud_api import RegCloud
api = RegCloud()
print("РЕГИОНЫ")
for r in api.regions():
print(f" {r['slug']:18s} {r['name']}")
print("\nСЕРВЕРЫ")
for s in api.servers():
info = api.size_info(s.get("size_slug", ""))
print(f" {s['id']} {s['name']:20s} {s['status']:8s} {RegCloud.ip_of(s):16s} "
f"{info.get('vcpus','?')} ядер, {info.get('memory_gb','?')} ГБ")
print("\nСНИМКИ")
for s in api.snapshots():
print(f" {s['id']} {s['name']:20s} {s.get('size_gigabytes')} ГБ")
print("\nКЛЮЧИ")
for k in api.ssh_keys():
print(f" {k['id']} {k['name']:15s} {k['fingerprint']}")
print("\nДЕНЬГИ")
for k, v in api.burn_rate().items():
print(f" {k}: {v}")
examples/provision.py
+31
Просмотреть файл
@@ -0,0 +1,31 @@
"""Развернуть сервер под клиента: выбрать регион и систему, поднять, отдать доступы.
Пример под перепродажу хостинга. Тариф собирается из требований, а не берётся
из готового списка, так что можно продавать любую конфигурацию.
"""
import sys
from regcloud_api import RegCloud, ValidationError
api = RegCloud()
CPU, RAM, DISK, LINE = 2, 4, 60, "hp"
REGION = "openstack-msk1"
OS = "ubuntu-24-04-amd64"
size = RegCloud.make_size(CPU, RAM, DISK, LINE) # проверит сочетание сам
print("тариф:", size, api.size_info(size))
available = {i.get("slug") for i in api.distributions()}
if OS not in available:
print(f"системы {OS} нет, доступны:", ", ".join(sorted(available)[:8]))
sys.exit(1)
try:
srv = api.create_server(size=size, image=OS, name=api.random_name(), region=REGION)
except ValidationError as exc:
print("платформа не приняла заказ:", exc)
sys.exit(1)
info = api.wait_active(srv["id"])
print(f"готово. сервер {info['id']}, адрес {RegCloud.ip_of(info)}")
print("пароль root приходит на почту, либо задайте ssh-ключ при создании")
regcloud_api/__init__.py
+18 -4
Просмотреть файл
@@ -1,5 +1,19 @@
"""Клиент к API облака REG.RU."""
from .client import RegCloud, load_state, save_state
"""Клиент к API облака REG.RU (Рег.облако).
__all__ = ["RegCloud", "load_state", "save_state"]
__version__ = "0.1.0"
Полное покрытие API: серверы, действия над ними, снимки, образы, тарифы, регионы,
SSH-ключи, IP-адреса, приватные сети, kubernetes, история операций и биллинг.
"""
from .client import RegCloud
from .errors import (AuthError, InsufficientFunds, Locked, NotFound, RegCloudError,
Timeout, ValidationError)
from .slugs import (ACTIONS, GPU_SIZES, IMAGE_TYPES, PLAN_LINES, REGIONS,
build_size, gpu_catalog, parse_size)
__all__ = [
"RegCloud",
"RegCloudError", "AuthError", "NotFound", "Locked", "ValidationError",
"InsufficientFunds", "Timeout",
"REGIONS", "PLAN_LINES", "GPU_SIZES", "IMAGE_TYPES", "ACTIONS",
"build_size", "parse_size", "gpu_catalog",
]
__version__ = "1.0.0"
regcloud_api/client.py
+447 -192
Просмотреть файл
@@ -1,22 +1,23 @@
"""Клиент к API облака REG.RU (Рег.облако).
API устроено как у DigitalOcean, сервер там называется reglet.
Токен берётся в личном кабинете, в разделе настроек.
Покрывает всё API целиком: серверы, действия над ними, снимки, образы, тарифы,
регионы, SSH-ключи, IP-адреса, приватные сети, историю операций, биллинг.
from regcloud_api import RegCloud
api = RegCloud() # токен из REGRU_CLOUD_TOKEN
for s in api.list():
print(s["id"], s["name"], s["status"])
api = RegCloud() # токен из REGRU_CLOUD_TOKEN
Зачем обёртка, если API простое. Затем, что у него есть неочевидные места,
собранные на живых серверах и обойдённые тут.
for r in api.regions(): # где можно размещаться
print(r["slug"], r["name"])
Адрес сервера лежит в поле ip, а не в networks, оно всегда пустое.
Снимок снимается только с работающего сервера, останавливает его платформа сама по offline.
Сразу после снимка сервер сидит в locked и удаляться отказывается примерно полминуты,
поэтому удаление настойчивое, иначе забытый сервер продолжит тарифицироваться.
Метод sizes возвращает не все линейки тарифов, полный список смотрите в панели управления.
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
@@ -24,204 +25,458 @@ import json
import os
import pathlib
import time
from typing import Any, Iterable
import httpx
BASE = "https://api.cloudvps.reg.ru/v1"
STATE_FILE = pathlib.Path(os.environ.get("REGCLOUD_STATE_DIR", ".")) / "regcloud_state.json"
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:
def __init__(self, token: str | None = None, timeout: float = 60.0):
self.token = token or os.environ.get("REGRU_CLOUD_TOKEN", "")
"""Всё 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 RuntimeError("нет токена, положи REGRU_CLOUD_TOKEN в окружение")
self.http = httpx.Client(
base_url=BASE, timeout=timeout,
headers={"Authorization": f"Bearer {self.token}", "Content-Type": "application/json"},
)
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 sizes(self) -> list[dict]:
r = self.http.get("/sizes")
r.raise_for_status()
return r.json().get("sizes", [])
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]:
"""Отбор тарифов с видеокартой, если они вам нужны.
Учтите, что 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
"""Тарифы с видеокартой. Собственные конфигурации для них платформа
не принимает, только эти готовые."""
return gpu_catalog()
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} секунд")
def size_info(self, slug: str) -> dict:
"""Разобрать slug тарифа на понятные значения."""
return parse_size(slug)
@staticmethod
def ip_of(reglet: dict) -> str:
"""Адрес лежит прямо в поле ip, а networks у этого провайдера пустует."""
if reglet.get("ip"):
return reglet["ip"]
nets = (reglet.get("networks") or {}).get("v4") or []
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})
# ---------- приватные сети ----------
def vpcs(self) -> list[dict]:
data = self.get("/vpcs")
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)
def create_vpc(self, name: str, region: str, members: list[int] | None = None) -> dict:
payload = {"name": name, "region_slug": region}
if members:
payload["members"] = members
data = self.post("/vpcs", json=payload)
return data.get("vpc", data)
def delete_vpc(self, vpc_id: int) -> bool:
self.delete(f"/vpcs/{vpc_id}")
return True
def vpc_members(self, vpc_id: int) -> list[dict]:
data = self.get(f"/vpcs/{vpc_id}/members")
return data.get("members", []) if isinstance(data, dict) else []
def vpc_attach(self, vpc_id: int, server_id: int) -> dict:
return self.post(f"/vpcs/{vpc_id}/members", json={"resource_id": server_id})
def vpc_detach(self, vpc_id: int, server_id: int) -> bool:
self.delete(f"/vpcs/{vpc_id}/members/{server_id}")
return True
# ---------- 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 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")
def __enter__(self): return self
def __exit__(self, *a): self.close()
regcloud_api/errors.py
+36
Просмотреть файл
@@ -0,0 +1,36 @@
"""Ошибки клиента. Разделены по смыслу, чтобы их можно было ловить прицельно."""
from __future__ import annotations
class RegCloudError(RuntimeError):
"""Общая ошибка платформы."""
def __init__(self, message: str, status: int = 0, code: str = "", payload: dict | None = None):
super().__init__(message)
self.status = status
self.code = code
self.payload = payload or {}
class AuthError(RegCloudError):
"""Токен не принят."""
class NotFound(RegCloudError):
"""Ресурса нет."""
class Locked(RegCloudError):
"""Ресурс занят другой операцией. Обычно проходит само за полминуты."""
class ValidationError(RegCloudError):
"""Платформа не приняла параметры. В payload лежит её объяснение."""
class InsufficientFunds(RegCloudError):
"""Не хватает денег на счету."""
class Timeout(RegCloudError):
"""Не дождались нужного состояния."""
regcloud_api/slugs.py
+134
Просмотреть файл
@@ -0,0 +1,134 @@
"""Справочники платформы и сборка slug'ов.
Часть этих данных API не отдаёт вовсе, часть отдаёт неполно, поэтому держим
их тут. Проверено 23.08.2026, при расхождении верить панели управления,
раздел настроек, пункт со slug'ами для API и Terraform.
"""
from __future__ import annotations
# Регионы. Список подтверждён самой платформой в тексте ошибки валидации.
REGIONS = {
"openstack-msk1": "Москва",
"openstack-msk2": "Москва-2",
"openstack-msk3": "Москва-3",
"openstack-spb1": "Санкт-Петербург",
"openstack-sam1": "Самара",
"openstack-fz1": "Москва ФЗ-152",
"msk1": "Москва, архивные тарифы",
}
# Линейки обычных тарифов
PLAN_LINES = {
"std": "Стандартный",
"stdp": "Стандартный плюс",
"hp": "Производительный",
"hps": "Производительный ФЗ-152",
"ddc": "Выделенный CPU",
"high": "Высокочастотный",
}
# Допустимые сочетания ядер и памяти по линейкам
CPU_RAM = {
"std": {1: [1, 2, 3, 4], 2: [2, 3, 6, 8], 3: [3, 4, 6, 8, 10],
4: [4, 6, 8, 10, 12, 16], 6: [6, 8, 10, 12, 16],
8: [8, 10, 12, 16, 20, 24, 32], 10: [10, 12, 16, 20, 24, 32, 40],
12: [12, 16, 20, 24, 32, 40, 48], 16: [16, 20, 24, 32, 40, 48],
20: [20, 24, 32, 40, 48], 24: [24, 32, 40, 48]},
"hp": {1: [1, 2, 3, 4], 2: [2, 3, 6, 8], 3: [3, 4, 6, 8, 10],
4: [4, 6, 8, 10, 12, 16], 6: [6, 8, 10, 12, 16],
8: [8, 10, 12, 16, 20, 24, 32], 10: [10, 12, 16, 20, 24, 32, 40],
12: [12, 16, 20, 24, 32, 40, 48], 16: [16, 20, 24, 32, 40, 48],
20: [20, 24, 32, 40, 48, 56, 64], 24: [24, 32, 40, 48, 56, 64, 72],
32: [32, 40, 48, 56, 64, 72, 80], 40: [40, 48, 56, 64, 72, 80],
48: [48, 56, 64, 72, 80]},
"high": {1: [1, 2, 3, 4], 2: [2, 3, 6, 8], 3: [3, 4, 6, 8, 10],
4: [4, 6, 8, 10, 12, 16], 6: [6, 8, 10, 12, 16],
8: [8, 10, 12, 16, 20, 24, 32], 10: [10, 12, 16, 20, 24, 32],
12: [12, 16, 20, 24, 32]},
}
CPU_RAM["stdp"] = CPU_RAM["std"]
CPU_RAM["hps"] = CPU_RAM["hp"]
CPU_RAM["ddc"] = CPU_RAM["hp"]
# Диапазон диска по линейкам, гигабайты, шаг десять
DISK_RANGE = {
"std": (10, 480), "stdp": (10, 480),
"hp": (10, 960), "hps": (10, 960), "ddc": (10, 960),
"high": (20, 320),
}
# Тарифы с видеокартой. Свои конфигурации для них собирать нельзя,
# только эти готовые. Цены в рублях за час на 23.08.2026.
GPU_SIZES = {
"c8-m32-d128-g1-a4000-gpu": ("A4000 16 ГБ", 8, 32, 128, 37.80),
"c16-m64-d256-g1-a4000-gpu": ("A4000 16 ГБ", 16, 64, 256, 51.87),
"c16-m64-d256-g1-a5000-gpu": ("A5000 24 ГБ", 16, 64, 256, 57.00),
"c24-m128-d512-g1-a5000-gpu": ("A5000 24 ГБ", 24, 128, 512, 82.79),
"c24-m128-d512-g2-a5000-gpu": ("2x A5000 24 ГБ", 24, 128, 512, 111.65),
"c32-m256-d1024-g1-a5000-gpu": ("A5000 24 ГБ", 32, 256, 1024, 132.01),
"c48-m384-d1024-g2-a5000-gpu": ("2x A5000 24 ГБ", 48, 384, 1024, 203.30),
"c16-m128-d512-g1-a100-gpu": ("A100 80 ГБ", 16, 128, 512, 222.14),
"c32-m256-d1024-g1-a100-gpu": ("A100 80 ГБ", 32, 256, 1024, 273.72),
"c64-m512-d1024-g2-a100-gpu": ("2x A100 80 ГБ", 64, 512, 1024, 529.14),
"c24-m96-d384-g1-rtx6000-gpu": ("RTX 6000 Blackwell", 24, 96, 384, 205.07),
"c32-m128-d512-g1-rtx6000-gpu": ("RTX 6000 Blackwell", 32, 128, 512, 223.43),
"c64-m192-d1024-g1-rtx6000-gpu": ("RTX 6000 Blackwell", 64, 192, 1024, 271.15),
"c32-m128-d1024-g2-rtx6000-gpu": ("2x RTX 6000 Blackwell", 32, 128, 1024, 382.58),
}
# Типы образов в API
IMAGE_TYPES = ("distribution", "application", "snapshot", "backup", "restore")
# Действия над сервером
ACTIONS = ("start", "stop", "reboot", "rebuild", "password_reset", "resize",
"generate_vnc_link", "snapshot", "enable_backups", "disable_backups",
"restore", "clone", "resize_isp_license")
ISP_LICENSES = ("isp_lite6", "isp_pro6", "isp_host6")
def build_size(cpu: int, ram: int, disk: int, line: str = "hp") -> str:
"""Собирает slug тарифа и проверяет сочетание по правилам платформы.
build_size(4, 8, 120, "hp") -> 'c4-m8-d120-hp'
"""
if line not in PLAN_LINES:
raise ValueError(f"нет линейки {line}, есть {', '.join(PLAN_LINES)}")
allowed = CPU_RAM.get(line, {})
if cpu not in allowed:
raise ValueError(f"в линейке {line} нет варианта на {cpu} ядер, "
f"доступны {sorted(allowed)}")
if ram not in allowed[cpu]:
raise ValueError(f"при {cpu} ядрах в линейке {line} память может быть "
f"{allowed[cpu]}, а не {ram}")
low, high = DISK_RANGE[line]
if not (low <= disk <= high) or disk % 10:
raise ValueError(f"диск в линейке {line} от {low} до {high} ГБ с шагом 10, а не {disk}")
return f"c{cpu}-m{ram}-d{disk}-{line}"
def parse_size(slug: str) -> dict:
"""Разбирает slug обратно в понятные значения."""
if slug in GPU_SIZES:
name, cpu, ram, disk, price = GPU_SIZES[slug]
return {"slug": slug, "gpu": name, "vcpus": cpu, "memory_gb": ram,
"disk_gb": disk, "price_hour": price, "line": "gpu"}
parts = slug.split("-")
try:
return {
"slug": slug,
"vcpus": int(parts[0].lstrip("c")),
"memory_gb": int(parts[1].lstrip("m")),
"disk_gb": int(parts[2].lstrip("d")),
"line": parts[3] if len(parts) > 3 else "",
"gpu": None,
}
except (IndexError, ValueError):
return {"slug": slug}
def gpu_catalog() -> list[dict]:
"""Тарифы с видеокартой списком, отсортированные по цене."""
out = [{"slug": s, "gpu": v[0], "vcpus": v[1], "memory_gb": v[2],
"disk_gb": v[3], "price_hour": v[4]} for s, v in GPU_SIZES.items()]
return sorted(out, key=lambda x: x["price_hour"])