76 lines
2.8 KiB
Python
76 lines
2.8 KiB
Python
"""REST client for the Sprut SDN API.
|
|
|
|
Mirrors sprut_api_call() from the bash original. Token, dry-run flag and
|
|
base URL are fixed at construction (replacing the bash globals $token /
|
|
$DRY_RUN / $sprut_api_base read by every call site) and a single
|
|
requests.Session is reused across all calls for connection reuse -- a
|
|
behaviorally-neutral efficiency improvement over one curl process per call.
|
|
|
|
The only deliberate behavioral deviation from the bash original (confirmed
|
|
with the user): a 30s timeout is applied to every HTTP call. curl in the
|
|
original has no --max-time at all, so a hung Sprut endpoint would hang the
|
|
whole script indefinitely; a timeout here is treated exactly like a non-2xx
|
|
response (error message to stderr, exit 1), so no new failure mode is
|
|
introduced, just an upper bound on how long a stuck call can hang.
|
|
|
|
The Keystone token is never refreshed after construction -- this directly
|
|
carries over the "token not refreshed across STAGE 2-4" known limitation
|
|
from the bash original. Do not add retry-on-401 here.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
|
|
import requests
|
|
|
|
from .redact import redact_psk
|
|
|
|
|
|
class SprutClient:
|
|
def __init__(self, base_url: str, token: str, dry_run: bool, timeout: float = 30.0):
|
|
self.base_url = base_url.rstrip("/")
|
|
self.token = token
|
|
self.dry_run = dry_run
|
|
self.timeout = timeout
|
|
self._session = requests.Session()
|
|
self._session.headers.update(
|
|
{
|
|
"Content-Type": "application/json",
|
|
"X-Auth-Token": token,
|
|
"X-SDN": "SPRUT",
|
|
}
|
|
)
|
|
|
|
def call(self, method: str, path: str, data: dict | None = None) -> dict:
|
|
url = f"{self.base_url}{path}"
|
|
|
|
if self.dry_run and method != "GET":
|
|
print(f"[DRY-RUN] Would {method} {url}", file=sys.stderr)
|
|
if data:
|
|
print(f"[DRY-RUN] Body: {json.dumps(redact_psk(data), indent=2)}", file=sys.stderr)
|
|
return {}
|
|
|
|
try:
|
|
response = self._session.request(
|
|
method, url, json=data if data is not None else None, timeout=self.timeout
|
|
)
|
|
except requests.exceptions.RequestException as exc:
|
|
print(f"Error: Sprut API {method} {url} failed: {exc}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
if not 200 <= response.status_code < 300:
|
|
print(f"Error: Sprut API {method} {url} returned HTTP {response.status_code}", file=sys.stderr)
|
|
print(f"Response: {response.text}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
if not response.text:
|
|
return {}
|
|
return response.json()
|
|
|
|
def get(self, path: str) -> dict:
|
|
return self.call("GET", path)
|
|
|
|
def post(self, path: str, data: dict) -> dict:
|
|
return self.call("POST", path, data)
|