import time import requests from config import Config from error_handling import AppServiceError, classify_http_error RETRYABLE_STATUS = {408, 425, 429, 500, 502, 503, 504} if Config.TLS_INSECURE_SKIP_VERIFY: requests.packages.urllib3.disable_warnings() # type: ignore[attr-defined] def request_json( url, method="GET", headers=None, payload=None, max_retry=3, retry_delay_ms=500, provider="external", ): attempt = 0 headers = headers or {} while attempt <= max_retry: start = time.perf_counter() try: response = requests.request( method=method, url=url, headers=headers, json=payload, timeout=20, verify=not Config.TLS_INSECURE_SKIP_VERIFY, ) cost_ms = round((time.perf_counter() - start) * 1000) data = None if response.text: try: data = response.json() except ValueError: data = response.text if not response.ok: retryable = response.status_code in RETRYABLE_STATUS code = str((data or {}).get("code", response.status_code)) if isinstance(data, dict) else str(response.status_code) message = (data or {}).get("message", f"HTTP {response.status_code}") if isinstance(data, dict) else f"HTTP {response.status_code}" category = classify_http_error(response.status_code, code, retryable=retryable) raise AppServiceError( message, category=category, code=code, status=response.status_code, retryable=retryable, provider=provider, detail={"data": data, "costMs": cost_ms, "url": url}, ) return {"data": data, "status": response.status_code, "cost_ms": cost_ms} except requests.Timeout as error: retryable = True if not retryable or attempt == max_retry: raise AppServiceError( "request timeout", category="timeout", code="REQUEST_TIMEOUT", status=None, retryable=True, provider=provider, detail={"url": url, "reason": str(error)}, ) from error delay = (2**attempt) * retry_delay_ms / 1000.0 time.sleep(delay) attempt += 1 except requests.RequestException as error: retryable = True if attempt == max_retry: raise AppServiceError( "network request failed", category="network", code="NETWORK_ERROR", status=None, retryable=True, provider=provider, detail={"url": url, "reason": str(error)}, ) from error delay = (2**attempt) * retry_delay_ms / 1000.0 time.sleep(delay) attempt += 1 except AppServiceError as error: retryable = error.retryable if not retryable or attempt == max_retry: raise delay = (2**attempt) * retry_delay_ms / 1000.0 time.sleep(delay) attempt += 1