86 lines
5.4 KiB
Python
86 lines
5.4 KiB
Python
import time
|
|
|
|
import httpx
|
|
|
|
TRANSIENT_STATUS = (502, 503, 504) # reverse proxy / Hub restarting
|
|
BACKOFF_SECONDS = (2, 5, 10, 20, 30, 30, 30, 30, 30) # 10 attempts, ~3 minutes of outage tolerated mid-job
|
|
|
|
|
|
def _resilient(send):
|
|
"""Runs one idempotent HTTP call, retrying transport failures (connect/TLS/read/protocol errors) and proxy 502/503/504 with backoff, so a network blip in
|
|
the middle of a long job does not throw the whole job (and minutes of classification) away. Non-transient errors (4xx) are returned untouched."""
|
|
last = None
|
|
for attempt in range(len(BACKOFF_SECONDS) + 1):
|
|
try:
|
|
response = send()
|
|
if response.status_code not in TRANSIENT_STATUS or attempt == len(BACKOFF_SECONDS):
|
|
return response
|
|
last = None
|
|
except httpx.TransportError as error:
|
|
last = error
|
|
if attempt == len(BACKOFF_SECONDS):
|
|
raise
|
|
time.sleep(BACKOFF_SECONDS[attempt])
|
|
raise last # pragma: no cover
|
|
|
|
|
|
class HubClient:
|
|
def __init__(self, url, token, chunk_size=25): self.url, self.token, self.chunk_size = url.rstrip("/"), token, chunk_size
|
|
def _headers(self): return {"Authorization": f"Bearer {self.token}"}
|
|
def classify_batch(self, products):
|
|
results = []
|
|
for start in range(0, len(products), self.chunk_size):
|
|
chunk = products[start:start + self.chunk_size]
|
|
# Small chunks + generous timeout + retry with backoff on any transport error or proxy 5xx: a large first sync must survive
|
|
# a slow Hub, a reverse-proxy read timeout or a TLS/network blip instead of failing the whole job.
|
|
response = _resilient(lambda: httpx.post(f"{self.url}/v1/classify/batch", json={"products": chunk}, headers=self._headers(), timeout=120))
|
|
response.raise_for_status(); results.extend(response.json()["results"])
|
|
return results
|
|
def get_json(self, path):
|
|
response = httpx.get(f"{self.url}{path}", headers=self._headers(), timeout=30)
|
|
response.raise_for_status(); return response.json()
|
|
def identity(self): return self.get_json("/v1/agents/me/identity")
|
|
def authorizations(self): return self.get_json("/v1/agents/me/authorizations")
|
|
def resolve_product(self, payload):
|
|
"""Asks the Hub to resolve a NEW product (the Agent never contacts external providers). Bounded: the Hub answers within its own time budget."""
|
|
response = httpx.post(f"{self.url}/v1/products/resolve", json=payload, headers=self._headers(), timeout=20)
|
|
response.raise_for_status(); return response.json()
|
|
def pending_enrichment(self):
|
|
"""Codes the Hub flagged for a new evaluation (panel button, Reprocessar/Sincronizar, KB decision)."""
|
|
response = httpx.get(f"{self.url}/v1/products/enrichment/pending", headers=self._headers(), timeout=15)
|
|
response.raise_for_status(); return response.json().get("codes") or []
|
|
def report_enrichment(self, payload):
|
|
response = _resilient(lambda: httpx.post(f"{self.url}/v1/products/enrichment/applied", json=payload, headers=self._headers(), timeout=30))
|
|
response.raise_for_status(); return response.json()
|
|
def health(self):
|
|
response = httpx.get(f"{self.url}/health", timeout=10)
|
|
return response.status_code == 200
|
|
def heartbeat(self, status, schema_fingerprint=None, agent_version=None, last_job_id=None):
|
|
payload = {"status": status, "schema_fingerprint": schema_fingerprint, "agent_version": agent_version, "last_job_id": last_job_id}
|
|
response = httpx.post(f"{self.url}/v1/agents/heartbeat", json=payload, headers=self._headers(), timeout=15)
|
|
response.raise_for_status(); return response.json()
|
|
def register(self, payload):
|
|
response = httpx.post(f"{self.url}/v1/agents/register", json=payload, headers=self._headers(), timeout=15)
|
|
response.raise_for_status(); return response.json()
|
|
def heartbeat_metadata(self, payload):
|
|
response = httpx.post(f"{self.url}/v1/agents/heartbeat", json=payload, headers=self._headers(), timeout=15)
|
|
response.raise_for_status(); return response.json()
|
|
def event(self, payload):
|
|
response = httpx.post(f"{self.url}/v1/agents/events", json=payload, headers=self._headers(), timeout=15)
|
|
response.raise_for_status(); return response.json()
|
|
def next_job(self):
|
|
response = httpx.get(f"{self.url}/v1/jobs/next", headers=self._headers(), timeout=15)
|
|
response.raise_for_status(); return response.json()
|
|
def create_job(self, kind, payload=None):
|
|
response = httpx.post(f"{self.url}/v1/jobs", json={"kind": kind, "payload": payload or {}}, headers=self._headers(), timeout=15)
|
|
response.raise_for_status(); return response.json()
|
|
def get_job(self, job_id):
|
|
response = httpx.get(f"{self.url}/v1/jobs/{job_id}", headers=self._headers(), timeout=15)
|
|
response.raise_for_status(); return response.json()
|
|
def complete_job(self, job_id, result_summary):
|
|
response = _resilient(lambda: httpx.post(f"{self.url}/v1/jobs/{job_id}/complete", json={"result_summary": result_summary}, headers=self._headers(), timeout=30))
|
|
response.raise_for_status(); return response.json()
|
|
def fail_job(self, job_id, error):
|
|
response = _resilient(lambda: httpx.post(f"{self.url}/v1/jobs/{job_id}/fail", json={"error": str(error)}, headers=self._headers(), timeout=30))
|
|
response.raise_for_status(); return response.json()
|