POST JSON to an upstream tracedecay path; returns (status, body).
(path: str, payload: dict)
| 724 | |
| 725 | |
| 726 | def _post_upstream_json(path: str, payload: dict) -> tuple[int, Any]: |
| 727 | """POST JSON to an upstream tracedecay path; returns (status, body).""" |
| 728 | base = _upstream_base() |
| 729 | url = f"{base}{path}" |
| 730 | req = urllib.request.Request( |
| 731 | url, |
| 732 | data=json.dumps(payload).encode("utf-8"), |
| 733 | method="POST", |
| 734 | headers={"Content-Type": "application/json"}, |
| 735 | ) |
| 736 | try: |
| 737 | with urllib.request.urlopen(req, timeout=_PROXY_TIMEOUT_SECONDS) as resp: # noqa: S310 — loopback/configured upstream only |
| 738 | return resp.status, json.loads(resp.read().decode("utf-8")) |
| 739 | except urllib.error.HTTPError as exc: |
| 740 | try: |
| 741 | return exc.code, json.loads(exc.read().decode("utf-8")) |
| 742 | except Exception: |
| 743 | return exc.code, {"detail": str(exc)} |
| 744 | except Exception as exc: |
| 745 | raise HTTPException( |
| 746 | status_code=502, |
| 747 | detail=f"tracedecay dashboard unreachable: {exc}", |
| 748 | ) from exc |
| 749 | |
| 750 | |
| 751 | def _build_curation_clusters( |
no test coverage detected