"""Transport layer — push probe results to target platform API with retry logic."""
from __future__ import annotations
import time
from pathlib import Path
import httpx
from core.base_probe import ProbeResult
from core.config import TargetConfig
MAX_RETRIES = 3
RETRY_BACKOFF = [1, 2, 4] # seconds
INGEST_ENDPOINT = "/api/probes/ingest"
def _push_with_retry(client: httpx.Client, url: str, json_body: dict, headers: dict) -> httpx.Response:
"""POST with exponential backoff retry on transient failures."""
last_error = None
for attempt in range(MAX_RETRIES):
try:
resp = client.post(url, json=json_body, headers=headers, timeout=30)
if resp.status_code < 500: # Don't retry client errors (4xx)
return resp
last_error = f"HTTP {resp.status_code}: {resp.text[:200]}"
except (httpx.ConnectError, httpx.TimeoutException, httpx.ReadTimeout) as e:
last_error = str(e)
if attempt < MAX_RETRIES - 1:
time.sleep(RETRY_BACKOFF[attempt])
raise httpx.HTTPError(f"Failed after {MAX_RETRIES} retries: {last_error}")
[docs]
def push_probes(output_dir: Path, target: TargetConfig, scan: bool = True) -> dict:
"""Push all probe results to target platform's /api/probes/ingest endpoint.
This is the recommended push method. The platform converts probe JSON
into DDL/PL/SQL files, stores them in Metastore, and optionally scans.
"""
if not target.url:
return {"status": "skipped", "reason": "no target URL configured"}
headers: dict[str, str] = {"Content-Type": "application/json"}
if target.api_key:
headers["Authorization"] = f"Bearer {target.api_key}"
# Collect all probe results
probes = []
for f in sorted(output_dir.rglob("*.json")):
try:
import json
probes.append(json.loads(f.read_text()))
except Exception:
pass
if not probes:
return {"status": "empty", "reason": "no probe results found"}
with httpx.Client(verify=True) as client:
try:
resp = _push_with_retry(
client,
f"{target.url.rstrip('/')}{INGEST_ENDPOINT}?project={target.project}&scan={str(scan).lower()}",
probes,
headers,
)
if resp.status_code == 401:
return {"status": "error", "error": "Authentication failed"}
resp.raise_for_status()
return {"status": "pushed", "probes": len(probes), **resp.json()}
except Exception as e:
return {"status": "error", "error": str(e)}
[docs]
def push_directory(output_dir: Path, target: TargetConfig) -> list[dict]:
"""Push all JSON files from output directory to target platform Metastore."""
if not target.url:
return [{"status": "skipped", "reason": "no target URL configured"}]
results = []
headers: dict[str, str] = {"Content-Type": "application/json"}
if target.api_key:
headers["Authorization"] = f"Bearer {target.api_key}"
with httpx.Client(verify=True) as client:
for f in sorted(output_dir.rglob("*.json")):
rel = f.relative_to(output_dir)
minio_path = f"probes/{rel}"
try:
resp = _push_with_retry(
client,
f"{target.url.rstrip('/')}/api/storage/files/create",
{"path": str(minio_path), "content": f.read_text(), "project": target.project},
headers,
)
if resp.status_code == 401:
results.append({"status": "error", "path": str(minio_path), "error": "Authentication failed. Check api_key."})
elif resp.status_code >= 400:
results.append({"status": "error", "path": str(minio_path), "error": f"HTTP {resp.status_code}"})
else:
results.append({"status": "pushed", "path": str(minio_path)})
except Exception as e:
results.append({"status": "error", "path": str(minio_path), "error": str(e)})
return results