1"""Measured benchmarks (no fabricated numbers): API lookup latency, job throughput, page response.2 AKOS_URL=… AKOS_TOKEN=… python scripts/bench.py [--n 50] [--jobs 20]"""3import argparse, json, os, statistics, sys, time4sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))5from akos.sdk import Akos6ap = argparse.ArgumentParser(); ap.add_argument("--n", type=int, default=50); ap.add_argument("--jobs", type=int, default=20); a = ap.parse_args()7A = Akos()8910def timed(fn, n):11 ts = []12 for _ in range(n):13 t0 = time.perf_counter(); fn(); ts.append((time.perf_counter() - t0) * 1000)14 return {"n": n, "p50_ms": round(statistics.median(ts), 1), "p95_ms": round(sorted(ts)[int(0.95 * (n - 1))], 1), "max_ms": round(max(ts), 1)}151617out = {"target": A.url, "when": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}18out["search_lookup"] = timed(lambda: A.search("exchange rates", 10), a.n)19pk = A.packages()[0]["id"]20out["akp_inspect"] = timed(lambda: A.package(pk), a.n)21out["overview_page_api"] = timed(lambda: A.status(), a.n)22import httpx23out["ui_index_html"] = timed(lambda: httpx.get(A.url + "/", timeout=30).raise_for_status(), a.n)24out["health"] = timed(lambda: A.health(), a.n)25t0 = time.perf_counter()26ids = [A._req("POST", "/jobs/enqueue-bench", json={}) if False else None for _ in range(0)]27# job throughput: enqueue N maintenance jobs through the ops path (admin) and wait for completion28import asyncio29jobs = []30for i in range(a.jobs):31 r = A._req("POST", "/ops/backup") if False else None32# use drift_check on a tested version as a representative worker job33v = next((p for p in A.packages() if p["lifecycle"] in ("tested", "registered", "monitored")), None)34if v:35 api = A.api(v["api_id"]); vid = api["current_version_id"]36 t0 = time.perf_counter()37 for i in range(a.jobs):38 jobs.append(A._req("POST", "/drift/check", json={"api_version_id": vid})["job_id"])39 for j in jobs:40 A.wait_job(j, timeout=600)41 el = time.perf_counter() - t042 out["job_throughput"] = {"jobs": a.jobs, "kind": "drift_check (live provider call + schema validation)", "elapsed_s": round(el, 2), "jobs_per_min": round(a.jobs / el * 60, 1)}43print(json.dumps(out, indent=1))44