generated from coulomb/repo-seed
fix(dashboard): enrich repo-sync page with live SBOM snapshot stats
repos.json.py now fetches /sbom/snapshots/ alongside /repos/ and annotates each repo with sbom_snapshot_count, sbom_entry_count, and a last_sbom_at fallback derived from actual snapshot data. This prevents "LastSBOM=never" when the denormalized field is out of sync. repo-sync.md gains SBOM KPI tiles (ingested vs no-SBOM), color-coded SBOM age column (same green/orange/red scale as state sync), and an entry count column showing packages from the latest snapshot. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -1,5 +1,5 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Observable data loader: fetches /repos/ from the API."""
|
||||
"""Observable data loader: fetches /repos/ enriched with SBOM snapshot stats."""
|
||||
import json
|
||||
import os
|
||||
import urllib.request
|
||||
@@ -7,9 +7,36 @@ import urllib.error
|
||||
|
||||
API_BASE = os.environ.get("API_BASE", "http://127.0.0.1:8000").rstrip("/")
|
||||
|
||||
try:
|
||||
with urllib.request.urlopen(f"{API_BASE}/repos/", timeout=10) as resp:
|
||||
data = json.loads(resp.read())
|
||||
print(json.dumps(data))
|
||||
except urllib.error.URLError as e:
|
||||
print(json.dumps({"error": str(e), "repos": []}))
|
||||
|
||||
def fetch(url):
|
||||
try:
|
||||
with urllib.request.urlopen(url, timeout=10) as resp:
|
||||
return json.loads(resp.read())
|
||||
except urllib.error.URLError as e:
|
||||
print(f"Warning: could not fetch {url}: {e}", flush=True)
|
||||
return []
|
||||
|
||||
|
||||
repos = fetch(f"{API_BASE}/repos/")
|
||||
snapshots = fetch(f"{API_BASE}/sbom/snapshots/")
|
||||
|
||||
# Build map: repo_id → {count, latest_at, latest_entry_count}
|
||||
snap_stats: dict = {}
|
||||
for s in snapshots:
|
||||
rid = s["repo_id"]
|
||||
if rid not in snap_stats:
|
||||
snap_stats[rid] = {"count": 0, "latest_at": None, "latest_entry_count": 0}
|
||||
snap_stats[rid]["count"] += 1
|
||||
if snap_stats[rid]["latest_at"] is None or s["snapshot_at"] > snap_stats[rid]["latest_at"]:
|
||||
snap_stats[rid]["latest_at"] = s["snapshot_at"]
|
||||
snap_stats[rid]["latest_entry_count"] = s["entry_count"]
|
||||
|
||||
# Enrich repos — fall back to snapshot data if denormalized field is missing
|
||||
for r in repos:
|
||||
stats = snap_stats.get(str(r["id"]), {})
|
||||
if not r.get("last_sbom_at") and stats.get("latest_at"):
|
||||
r["last_sbom_at"] = stats["latest_at"]
|
||||
r["sbom_snapshot_count"] = stats.get("count", 0)
|
||||
r["sbom_entry_count"] = stats.get("latest_entry_count", 0)
|
||||
|
||||
print(json.dumps(repos))
|
||||
|
||||
Reference in New Issue
Block a user