curl http://127.0.0.1:8005/stats
{ "total_jobs": 150, "by_status": { "queued": 5, "running": 3, "succeeded": 130, "failed": 10, "cancelled": 2 } }
Get statistics about async jobs
import httpx import time API_URL = "http://127.0.0.1:8005" FAILURE_THRESHOLD = 0.1 # 10% failure rate def check_failure_rate(): stats = httpx.get(f"{API_URL}/stats").json() total = stats["total_jobs"] failed = stats["by_status"].get("failed", 0) if total > 0: failure_rate = failed / total if failure_rate > FAILURE_THRESHOLD: print(f"WARNING: High failure rate: {failure_rate:.1%}") return False print(f"OK: {total} jobs, {failed} failed ({failed/max(total,1):.1%})") return True # Periodic monitoring while True: check_failure_rate() time.sleep(60)