tegwick
5fc4c56215
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a028f0-a42f-7582-89a8-ebaad7343834
56 lines
1.8 KiB
Python
56 lines
1.8 KiB
Python
"""Read-only client for the reversible SBOM Nexus compatibility facade."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from api.config import settings
|
|
|
|
|
|
class SBOMNexusError(RuntimeError):
|
|
"""An SBOM Nexus request failed without a usable compatibility response."""
|
|
|
|
def __init__(self, status_code: int, detail: str):
|
|
super().__init__(detail)
|
|
self.status_code = status_code
|
|
self.detail = detail
|
|
|
|
|
|
def reads_from_nexus() -> bool:
|
|
return settings.sbom_nexus_read_mode == "nexus"
|
|
|
|
|
|
async def get_json(
|
|
path: str,
|
|
*,
|
|
params: dict[str, Any] | None = None,
|
|
) -> Any:
|
|
"""Fetch one Nexus resource; never fall back silently to legacy storage."""
|
|
if not settings.sbom_nexus_url:
|
|
raise SBOMNexusError(503, "SBOM Nexus read mode is enabled without SBOM_NEXUS_URL")
|
|
|
|
try:
|
|
async with httpx.AsyncClient(
|
|
base_url=settings.sbom_nexus_url.rstrip("/"),
|
|
timeout=settings.sbom_nexus_timeout_seconds,
|
|
) as client:
|
|
response = await client.get(path, params=params)
|
|
except httpx.RequestError as exc:
|
|
raise SBOMNexusError(502, f"SBOM Nexus is unavailable: {exc.__class__.__name__}") from exc
|
|
|
|
if response.status_code >= 400:
|
|
detail = f"SBOM Nexus returned HTTP {response.status_code}"
|
|
try:
|
|
payload = response.json()
|
|
if isinstance(payload, dict) and isinstance(payload.get("detail"), str):
|
|
detail = payload["detail"]
|
|
except ValueError:
|
|
pass
|
|
status_code = 404 if response.status_code == 404 else 502
|
|
raise SBOMNexusError(status_code, detail)
|
|
|
|
try:
|
|
return response.json()
|
|
except ValueError as exc:
|
|
raise SBOMNexusError(502, "SBOM Nexus returned invalid JSON") from exc
|