Skip to main content
GET
/
stats
curl http://127.0.0.1:8005/stats

import 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
  }
}
Get aggregate statistics about all async jobs.

​
Response

​
total_jobs
integer
required
Total number of jobs ever submitted.
​
by_status
object
required
Breakdown of jobs by status.
​
by_status.queued
integer
Number of jobs currently queued.
​
by_status.running
integer
Number of jobs currently running.
​
by_status.succeeded
integer
Number of jobs that completed successfully.
​
by_status.failed
integer
Number of jobs that failed.
​
by_status.cancelled
integer
Number of jobs that were cancelled.
curl http://127.0.0.1:8005/stats

import 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)

​
See Also