tegwick
b75234a533
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a028f0-a42f-7582-89a8-ebaad7343834
70 lines
2.2 KiB
Python
70 lines
2.2 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"
|
|
|
|
|
|
def writes_to_nexus() -> bool:
|
|
return settings.sbom_nexus_write_mode == "nexus"
|
|
|
|
|
|
async def request_json(
|
|
method: str,
|
|
path: str,
|
|
*,
|
|
params: dict[str, Any] | None = None,
|
|
body: dict[str, Any] | None = None,
|
|
) -> Any:
|
|
"""Call Nexus; 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.request(method, path, params=params, json=body)
|
|
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
|
|
|
|
|
|
async def get_json(path: str, *, params: dict[str, Any] | None = None) -> Any:
|
|
return await request_json("GET", path, params=params)
|
|
|
|
|
|
async def post_json(path: str, *, body: dict[str, Any]) -> Any:
|
|
return await request_json("POST", path, body=body)
|