Stats API

Get statistics about async jobs

Get aggregate statistics about all async jobs.

Response

ResponseField integer

Total number of jobs ever submitted.

ResponseField object

Breakdown of jobs by status.

ResponseField integer

Number of jobs currently queued.

ResponseField integer

Number of jobs currently running.

ResponseField integer

Number of jobs that completed successfully.

ResponseField integer

Number of jobs that failed.

ResponseField integer

Number of jobs that were cancelled.

cURL
curl http://127.0.0.1:8005/stats
Python
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)}")
CLI
curl http://127.0.0.1:8005/stats
200 OK
{
  "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

python
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