54 lines
2.2 KiB
Python
54 lines
2.2 KiB
Python
from __future__ import annotations
|
|
import requests
|
|
from requests.adapters import HTTPAdapter
|
|
from urllib3.util.retry import Retry
|
|
|
|
DEFAULT_TIMEOUT = 10.0
|
|
|
|
class ComboCoreClient:
|
|
def __init__(self, host: str, username: str, password: str, verify_ssl: bool = False, timeout: float = DEFAULT_TIMEOUT):
|
|
self.base = f"https://{host}"
|
|
self.username = username
|
|
self.password = password
|
|
self.verify = verify_ssl
|
|
self.timeout = timeout
|
|
self._token = None
|
|
self._s = requests.Session()
|
|
retry = Retry(total=3, backoff_factor=0.3, status_forcelist=[429,500,502,503,504])
|
|
self._s.mount("https://", HTTPAdapter(max_retries=retry))
|
|
self._s.mount("http://", HTTPAdapter(max_retries=retry))
|
|
|
|
# ----- PLS auth -----
|
|
def login(self) -> str:
|
|
r = self._s.post(
|
|
f"{self.base}/core/pls/api/1/auth/login",
|
|
json={"username": self.username, "password": self.password},
|
|
timeout=self.timeout, verify=self.verify
|
|
)
|
|
r.raise_for_status()
|
|
j = r.json()
|
|
self._token = j.get("access_token") or j.get("token")
|
|
if not self._token:
|
|
raise RuntimeError("No access token in login response.")
|
|
return self._token
|
|
|
|
def _auth_headers(self):
|
|
if not self._token:
|
|
self.login()
|
|
return {"Authorization": f"Bearer {self._token}"}
|
|
|
|
def get(self, path: str):
|
|
r = self._s.get(f"{self.base}{path}", headers=self._auth_headers(), timeout=self.timeout, verify=self.verify)
|
|
if r.status_code == 401:
|
|
self._token = None
|
|
r = self._s.get(f"{self.base}{path}", headers=self._auth_headers(), timeout=self.timeout, verify=self.verify)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
def post(self, path: str, json=None):
|
|
r = self._s.post(f"{self.base}{path}", headers=self._auth_headers(), json=json or {}, timeout=self.timeout, verify=self.verify)
|
|
if r.status_code == 401:
|
|
self._token = None
|
|
r = self._s.post(f"{self.base}{path}", headers=self._auth_headers(), json=json or {}, timeout=self.timeout, verify=self.verify)
|
|
r.raise_for_status()
|
|
return r.json() |