GET an upstream tracedecay path and return the parsed JSON object.
(path: str)
| 699 | |
| 700 | |
| 701 | def _get_upstream_json(path: str) -> dict: |
| 702 | """GET an upstream tracedecay path and return the parsed JSON object.""" |
| 703 | base = _upstream_base() |
| 704 | url = f"{base}{path}" |
| 705 | req = urllib.request.Request(url, method="GET") |
| 706 | try: |
| 707 | with urllib.request.urlopen(req, timeout=_PROXY_TIMEOUT_SECONDS) as resp: # noqa: S310 — loopback/configured upstream only |
| 708 | payload = json.loads(resp.read().decode("utf-8")) |
| 709 | except urllib.error.HTTPError as exc: |
| 710 | raise HTTPException( |
| 711 | status_code=502, |
| 712 | detail=f"tracedecay dashboard returned {exc.code} for {path}", |
| 713 | ) from exc |
| 714 | except Exception as exc: |
| 715 | raise HTTPException( |
| 716 | status_code=502, |
| 717 | detail=f"tracedecay dashboard unreachable: {exc}", |
| 718 | ) from exc |
| 719 | if not isinstance(payload, dict): |
| 720 | raise HTTPException( |
| 721 | status_code=502, detail=f"unexpected upstream payload for {path}" |
| 722 | ) |
| 723 | return payload |
| 724 | |
| 725 | |
| 726 | def _post_upstream_json(path: str, payload: dict) -> tuple[int, Any]: |
no test coverage detected