tegwick
4642a53d6e
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>
43 lines
1.4 KiB
Python
43 lines
1.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Observable data loader: fetches /repos/ enriched with SBOM snapshot stats."""
|
|
import json
|
|
import os
|
|
import urllib.request
|
|
import urllib.error
|
|
|
|
API_BASE = os.environ.get("API_BASE", "http://127.0.0.1:8000").rstrip("/")
|
|
|
|
|
|
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))
|