(endpoint: str, payload: Dict[str, Any], timeout: int = 300)
| 68 | |
| 69 | |
| 70 | def _api(endpoint: str, payload: Dict[str, Any], timeout: int = 300): |
| 71 | assert VLLM_API_URL, "VLLM API URL not set" |
| 72 | url = f"{VLLM_API_URL}/v1/{endpoint}" |
| 73 | for attempt in range(3): |
| 74 | try: |
| 75 | logging.debug("POST %s try %d payload %s", endpoint, attempt + 1, payload) |
| 76 | r = API.post(url, json=payload, timeout=timeout) |
| 77 | if r.status_code == 200: |
| 78 | if r.headers.get("Content-Type", "").startswith("application/json"): |
| 79 | return r.json() |
| 80 | return r.text or True |
| 81 | r.raise_for_status() |
| 82 | except Exception as e: |
| 83 | logging.warning("API error %s - attempt %d/3", e, attempt + 1) |
| 84 | time.sleep(2 * (attempt + 1)) |
| 85 | logging.error("API %s failed after retries", endpoint) |
| 86 | return None |
| 87 | |
| 88 | |
| 89 | def load_adapter(path: str, name: str) -> bool: |
no outgoing calls
no test coverage detected