Observability

quiv gives you a snapshot of the scheduler at one moment, and queries over tasks and jobs that accept filters. Together they are enough to build an admin page or a health check without reading the database directly.

stats()

stats() returns a frozen QuivStats dataclass:

from quiv import Quiv, QuivStats

stats: QuivStats = scheduler.stats()
stats.active_jobs        # jobs currently executing
stats.pool_size          # thread-pool size
stats.pool_utilization   # active_jobs / pool_size, 0.0-1.0
stats.tasks_by_status    # e.g. {"active": 3, "paused": 1}
stats.next_run_at        # earliest upcoming run (UTC), or None
stats.job_history_count  # job rows currently retained
stats.loop_alive         # same value as scheduler.is_running

QuivStats is a plain dataclass. Call dataclasses.asdict() on it to build a JSON response.

Health

is_running answers one question: does the scheduler loop run? It is True when start() was called, shutdown() was not, and the loop thread is alive.

from fastapi.responses import JSONResponse


@app.get("/health")
def health() -> JSONResponse:
    if not scheduler.is_running:
        return JSONResponse({"scheduler": "stopped"}, status_code=503)
    return JSONResponse({"scheduler": "running"})

stats().loop_alive holds the same value. stats() reads the database and is_running does not, so use is_running in a health probe. Running in a Container shows a Docker HEALTHCHECK for this endpoint.

Job queries

get_all_jobs() accepts filters, an order, and paging:

scheduler.get_all_jobs(
    status=JobStatus.FAILED,   # optional status filter
    task_id=task_id,           # only this task's jobs
    since=window_start,        # started_at >= since (UTC)
    until=window_end,          # started_at <= until (UTC)
    order_by="started_at",     # "started_at" | "ended_at"
    descending=True,           # newest first by default
    limit=20,
    offset=0,
)

order_by accepts "started_at" and "ended_at". Any other value raises ConfigurationError. The two datetime filters follow the rule for run_at: quiv reads a naive value as UTC, and converts an aware value to UTC.

Task queries

get_all_tasks() accepts status, task_name, limit, and offset, next to include_run_once. It orders the results by next_run_at, earliest first.

scheduler.get_all_tasks(status=TaskStatus.PAUSED)
scheduler.get_all_tasks(limit=50, offset=100)
scheduler.get_all_tasks(task_name="nightly-sync", include_run_once=True)

task_name is an exact match. Names can repeat, so the list can hold more than one task. Pass include_run_once=True to find a one-off by its name.

Outstanding main-loop work

pending_main_loop_work() returns how many callables handed to the main loop by run_on_main have not finished.

scheduler.pending_main_loop_work()   # -> int

quiv never waits for that work and never cancels it, because the loop is your application's. This is the number that lets you decide when the loop may close:

scheduler.shutdown()
while scheduler.pending_main_loop_work():
    await asyncio.sleep(0.05)

Unlike stats(), it reads one set under a lock and never touches the database, so it is cheap to poll.

It can read one higher than the number of handoffs for a few microseconds, while a callable is counted both as the marker that reserved its place and as the task or future that replaced it. It never reads lower than the truth, which is the direction that matters: a count that briefly missed work already running would tell you the loop was safe to close when it was not.

Example endpoints

The FastAPI example app puts all three into routes:

@router.get("/stats")
def get_stats():
    return asdict(scheduler.stats())


@router.get("/{task_id}/jobs")
def list_task_jobs(task_id: str, limit: int | None = None, offset: int = 0):
    return scheduler.get_all_jobs(task_id=task_id, limit=limit, offset=offset)


@router.patch("/{task_id}")
def update_task(task_id: str, update: TaskUpdate):
    return scheduler.update_task(task_id, **update.model_dump(exclude_none=True))

See update_task() to change a task while the scheduler runs.