70 lines
2.1 KiB
Python
70 lines
2.1 KiB
Python
import requests
|
||
|
||
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."""
|
||
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):
|
||
"""dev/test по URL стенда."""
|
||
for name in ("dev", "test"):
|
||
if name in (endpoint or ""):
|
||
return name
|
||
return "?"
|
||
|
||
|
||
def create_client(token, fallback_endpoint=None):
|
||
"""HttpClient с автоопределением стенда по токену."""
|
||
ep = detect_endpoint(token) or fallback_endpoint
|
||
if not ep:
|
||
return None
|
||
return HttpClient(ep, token), ep
|
||
|
||
|
||
class HttpClient:
|
||
def __init__(self, endpoint, token):
|
||
self._endpoint = endpoint.rstrip("/")
|
||
self._session = requests.Session()
|
||
self._session.headers.update({
|
||
"Authorization": f"Bearer {token}",
|
||
"User-Agent": "Mozilla/5.0",
|
||
})
|
||
|
||
def get(self, path, **kwargs):
|
||
kwargs.setdefault("timeout", 10)
|
||
r = self._session.get(f"{self._endpoint}{path}", **kwargs)
|
||
r.raise_for_status()
|
||
return r.json()
|
||
|
||
def post(self, path, data=None, **kwargs):
|
||
kwargs.setdefault("timeout", 30)
|
||
url = f"{self._endpoint}{path}"
|
||
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 = {}
|
||
try:
|
||
parsed = r.json()
|
||
if isinstance(parsed, dict):
|
||
result = parsed
|
||
except Exception:
|
||
pass
|
||
loc = r.headers.get("Location", "")
|
||
if loc:
|
||
result["_location"] = loc
|
||
return result
|