Stats API
Get statistics about async jobs
Get aggregate statistics about all async jobs.
Response
ResponseField
integerTotal number of jobs ever submitted.
ResponseField
objectBreakdown of jobs by status.
ResponseField
integerNumber of jobs currently queued.
ResponseField
integerNumber of jobs currently running.
ResponseField
integerNumber of jobs that completed successfully.
ResponseField
integerNumber of jobs that failed.
ResponseField
integerNumber of jobs that were cancelled.
curl http://127.0.0.1:8005/statsimport httpx
response = httpx.get("http://127.0.0.1:8005/stats")
stats = response.json()
print(f"Total jobs: {stats['total_jobs']}")
print(f"Running: {stats['by_status'].get('running', 0)}")
print(f"Succeeded: {stats['by_status'].get('succeeded', 0)}")
print(f"Failed: {stats['by_status'].get('failed', 0)}")curl http://127.0.0.1:8005/stats{
"total_jobs": 150,
"by_status": {
"queued": 5,
"running": 3,
"succeeded": 130,
"failed": 10,
"cancelled": 2
}
}Usage
Use this endpoint for:
- Monitoring dashboards
- Capacity planning
- Alerting on failure rates
Example: Monitor Failure Rate
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)