Stats API

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

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

integer

required

Total number of jobs ever submitted.

object

required

Breakdown of jobs by status.

integer

Number of jobs currently queued.

integer

Number of jobs currently running.

integer

Number of jobs that completed successfully.

integer

Number of jobs that failed.

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)

Deploy

Response

Usage

Example: Monitor Failure Rate

See Also

​Response

​Usage

​Example: Monitor Failure Rate

​See Also

Response

Usage

Example: Monitor Failure Rate

See Also