186 lines
8.7 KiB
Python
186 lines
8.7 KiB
Python
"""
|
||
HTTP-клиент для Nubes API — тонкая обёртка над requests.Session.
|
||
|
||
HttpClient:
|
||
- Добавляет обязательные заголовки:
|
||
Authorization: Bearer <token> — аутентификация
|
||
User-Agent: Mozilla/5.0 — DDoS-Guard блокирует python-requests по умолчанию
|
||
- GET: raise_for_status → .json() — автоматически проверяет HTTP-статус
|
||
- POST: проверка r.ok, извлечение Location-заголовка + UUID
|
||
- raw_delete: DELETE без авторизации (для CMDB)
|
||
|
||
Функции автоопределения стенда:
|
||
- detect_endpoint(token) — пробует dev→test стенды по токену
|
||
- stand_name(endpoint) — "dev" / "test" по URL
|
||
- create_client(token) — HttpClient + endpoint одним вызовом
|
||
|
||
Зачем автоопределение: пользователь вводит токен, мы не знаем dev это или test.
|
||
Пробуем оба стенда — какой ответит с results != None, тот и рабочий.
|
||
"""
|
||
|
||
import requests
|
||
|
||
# Список стендов для автоопределения.
|
||
# Порядок ВАЖЕН: dev первый — он быстрее (меньше нагрузка), test — резервный.
|
||
STANDS = [
|
||
"https://lk-api-gateway-dev.ngcloud.ru/api/v1/svc",
|
||
"https://lk-api-gateway-test.ngcloud.ru/api/v1/svc",
|
||
]
|
||
|
||
|
||
def detect_endpoint(token):
|
||
"""Пробуем токен против dev и test стендов, возвращаем рабочий URL.
|
||
|
||
Алгоритм:
|
||
1. Для каждого стенда создаём HttpClient + делаем GET /instances?pageSize=1.
|
||
2. Если results != None — стенд рабочий, токен валиден → возвращаем URL.
|
||
3. Если Exception (401/403/таймаут) — пробуем следующий.
|
||
4. Ни один не подошёл → None.
|
||
|
||
Почему results != None а не просто HTTP 200:
|
||
API может вернуть 200 с пустым списком (нет инстансов) — это норм.
|
||
results=None — признак что ответ не соответствует ожидаемой структуре."""
|
||
for ep in STANDS:
|
||
try:
|
||
c = HttpClient(ep, token)
|
||
data = c.get("/instances", params={"pageSize": 1, "page": 1})
|
||
if data.get("results") is not None:
|
||
return ep
|
||
except Exception:
|
||
continue
|
||
return None
|
||
|
||
|
||
def stand_name(endpoint):
|
||
"""Определить имя стенда по URL.
|
||
|
||
Ищет подстроку 'dev' или 'test' в URL.
|
||
Если не найдено — '?' (неизвестный стенд)."""
|
||
for name in ("dev", "test"):
|
||
if name in (endpoint or ""):
|
||
return name
|
||
return "?"
|
||
|
||
|
||
def create_client(token, fallback_endpoint=None):
|
||
"""HttpClient с автоопределением стенда.
|
||
|
||
Сначала detect_endpoint(token) — dev→test.
|
||
Если не сработало — fallback_endpoint (из конфига).
|
||
Возвращает (HttpClient, endpoint_url) или None."""
|
||
ep = detect_endpoint(token) or fallback_endpoint
|
||
if not ep:
|
||
return None
|
||
return HttpClient(ep, token), ep
|
||
|
||
|
||
class HttpClient:
|
||
"""HTTP-клиент для Nubes REST API.
|
||
|
||
Использует requests.Session для:
|
||
- Keep-alive (переиспользование TCP+TLS между запросами)
|
||
- Единые заголовки для всех запросов
|
||
|
||
Обязательные заголовки:
|
||
- Authorization: Bearer <token> — без этого API вернёт 401
|
||
- User-Agent: Mozilla/5.0 — DDoS-Guard блокирует "python-requests/2.x"
|
||
|
||
Таймауты: GET=10с (лёгкие), POST=30с (создание ресурсов дольше)."""
|
||
|
||
def __init__(self, endpoint, token):
|
||
# Убираем trailing slash (".../svc/" → ".../svc")
|
||
# чтобы path добавлялся как "/path", а не "path"
|
||
self._endpoint = endpoint.rstrip("/")
|
||
# Session переиспользует TCP + TLS handshake между запросами.
|
||
# Без Session каждый запрос делал бы новый connect → медленно.
|
||
self._session = requests.Session()
|
||
self._session.headers.update({
|
||
"Authorization": f"Bearer {token}",
|
||
"User-Agent": "Mozilla/5.0", # DDoS-Guard: не python-requests
|
||
})
|
||
|
||
def get(self, path, **kwargs):
|
||
"""GET-запрос к Nubes API.
|
||
|
||
Особенности:
|
||
- Таймаут 10 секунд по умолчанию.
|
||
- raise_for_status() — 4xx/5xx → HTTPError (ловим выше).
|
||
- Пустой ответ (нет body) → {} (норма для validate-cfs).
|
||
|
||
Args:
|
||
path: str — путь относительно endpoint (напр. "/instances").
|
||
**kwargs — params, timeout, headers и т.д.
|
||
|
||
Returns:
|
||
dict/list — распарсенный JSON, или {} если ответ пустой."""
|
||
kwargs.setdefault("timeout", 10)
|
||
r = self._session.get(f"{self._endpoint}{path}", **kwargs)
|
||
r.raise_for_status()
|
||
if not r.text or not r.text.strip():
|
||
return {} # пустой ответ — норма для некоторых эндпоинтов
|
||
return r.json()
|
||
|
||
def post(self, path, data=None, **kwargs):
|
||
"""POST-запрос к Nubes API.
|
||
|
||
Особенности:
|
||
- Таймаут 30 секунд по умолчанию.
|
||
- data → json=... — requests сам ставит Content-Type: application/json.
|
||
- Извлекает Location-заголовок → UUID → кладёт в instanceUid/instanceOperationUid.
|
||
Это нужно чтобы не полагаться только на тело ответа (которое может быть пустым).
|
||
- При HTTP-ошибке: Exception с кодом и телом ответа.
|
||
|
||
Args:
|
||
path: str — путь (напр. "/instances").
|
||
data: dict|None — тело запроса (сериализуется в JSON).
|
||
|
||
Returns:
|
||
dict с полями ответа + _status + _location (+ UUID если найден)."""
|
||
kwargs.setdefault("timeout", 30)
|
||
url = f"{self._endpoint}{path}"
|
||
# json=... — requests сам сериализует и ставит Content-Type: application/json
|
||
r = self._session.post(url, json=(data if data is not None else {}), **kwargs)
|
||
if not r.ok:
|
||
raise Exception(f"POST {path}: {r.status_code} {r.reason}: {r.text[:200]}")
|
||
result = {"_status": r.status_code}
|
||
try:
|
||
parsed = r.json()
|
||
if isinstance(parsed, dict):
|
||
result = parsed
|
||
result["_status"] = r.status_code
|
||
except Exception:
|
||
pass # тело не JSON — ок, UUID извлечём из Location
|
||
# Location-заголовок: "./UUID" или "/api/v1/svc/.../UUID"
|
||
loc = r.headers.get("Location", "")
|
||
if loc:
|
||
result["_location"] = loc
|
||
# Извлекаем UUID — последний сегмент после split("/")
|
||
parts = loc.rstrip("/").split("/")
|
||
uid = parts[-1]
|
||
# Проверка: не ".", длина >= 32 (UUID = 36 символов с дефисами)
|
||
if uid and uid != "." and len(uid) >= 32:
|
||
# Кладём UUID в правильное поле в зависимости от эндпоинта
|
||
if "/instanceOperations" in path:
|
||
result["instanceOperationUid"] = uid
|
||
elif "/instances" in path:
|
||
result["instanceUid"] = uid
|
||
return result
|
||
|
||
def raw_delete(self, url):
|
||
"""DELETE-запрос к произвольному URL — БЕЗ авторизации.
|
||
|
||
Используется для CMDB API (cmdb-api.deck.nubes.ru) —
|
||
жёсткое удаление недосозданных/зависших инстансов.
|
||
CMDB не требует Bearer-токена.
|
||
|
||
Args:
|
||
url: str — полный URL (не path, другой хост!)
|
||
|
||
Returns:
|
||
(ok: bool, status_code: int|str)."""
|
||
try:
|
||
r = requests.delete(url, timeout=10)
|
||
return r.ok, r.status_code
|
||
except Exception as e:
|
||
return False, str(e)
|