# quiv — full documentation
> Concatenated from https://nandyalu.github.io/quiv for LLM consumption.
> Index: https://nandyalu.github.io/quiv/llms.txt
---
#
Background tasks for FastAPI apps that need more than `BackgroundTasks` and less than Celery.
If you've reached for APScheduler inside a FastAPI app, you've probably hit one of these:
- A task is running too long and the user wants to cancel it — but there's no clean way to signal the worker mid-run.
- A background job needs to push progress to a websocket, and you're writing `run_coroutine_threadsafe` glue to hop back onto the main loop.
- You want a job id stamped on every log line for one specific run, and you're threading it through call sites by hand.
- You have a complete async pipeline you want to run in the background, and you're wrapping it in `asyncio.run` just to hand it to a sync-only scheduler.
`quiv` was built inside [Trailarr](https://github.com/nandyalu/trailarr) — a FastAPI app that outgrew APScheduler for exactly these reasons. It's a single-process, threadpool-backed scheduler with first-class support for cooperative cancellation (`_stop_event`), main-loop progress callbacks (`_progress_hook`), and per-job tracing (`_job_id`).
It is not a Celery replacement. If you need multi-process workers, durable queues, or distributed execution, use Celery or arq. `quiv` is for the in-process case those tools are overkill for.
Supports Python 3.10 through 3.14.
## Install
=== "uv"
```bash
uv add quiv
```
=== "pip"
```bash
pip install quiv
```
## Quick example
A full FastAPI integration — lifespan startup, an endpoint that schedules work, and progress streaming back to the main loop:
```python
from contextlib import asynccontextmanager
from fastapi import FastAPI
from quiv import Quiv
# Create the Quiv scheduler
scheduler = Quiv(timezone="UTC")
# Wire it up in FastAPI's lifespan so that it starts and dies with your app
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup
scheduler.start()
yield
# Shutdown
scheduler.shutdown()
# Create FastAPI app
app = FastAPI(lifespan=lifespan)
# Create a test function that we can later schedule to broadcast progress
# sync/async - doesn't matter; quiv handles them all
def ping(_progress_hook=None):
for i in range(30):
# do some work
if _progress_hook:
_progress_hook(message="ping", progress=i, total=30)
# Now the actual progress callback function that we want to run on the main asyncio loop
async def on_progress(**payload):
# Replace with websocket broadcast, logging, metrics, etc.
print("progress", payload)
# Create the endpoint function that will schedule the task when triggered
@app.post("/start-heartbeat")
def start_heartbeat():
task_id = scheduler.add_task(
task_name="heartbeat",
func=ping,
interval=30,
progress_callback=on_progress,
)
return {"task_id": task_id}
```
## What you actually get
### Run async handlers natively, no `asyncio.run` wrapper
APScheduler has asyncio integrations, but async pipelines can still end up wrapped or bridged when you’re scheduling from a threadpool. `quiv` accepts async handlers directly; each invocation runs in an event loop created on the worker thread for that job. Sync and async handlers coexist in the same scheduler.
```python
async def fetch_updates(_stop_event=None):
await some_async_api_call()
scheduler.add_task(task_name="fetch", func=fetch_updates, interval=60)
```
### Cancel a running task from an HTTP endpoint
`_stop_event` is a per-job `threading.Event` injected into your handler. Check it at natural breakpoints and exit early when an endpoint calls `scheduler.cancel_job(job_id)` — no thread killing, no exceptions raised across thread boundaries.
```python
def download(media_id: int, _stop_event=None):
for chunk in stream_chunks(media_id):
if _stop_event and _stop_event.is_set():
return # cooperative exit
write(chunk)
```
### Stream progress to a websocket without the `run_coroutine_threadsafe` dance
Your handler calls `_progress_hook(**payload)` from inside the threadpool. `quiv` dispatches your registered async callback on the main asyncio loop — where it can broadcast over a websocket, update app state, or push to a metrics client.
```python
async def on_progress(**payload):
await websocket_manager.broadcast(payload) # runs on the main loop
scheduler.add_task(
task_name="download",
func=download,
progress_callback=on_progress,
run_once=True,
)
```
### Dispatch to the main loop from anywhere — no parameter threading
When a deeply-nested function inside a task needs to touch a resource that lives on the main event loop (e.g. a WebSocket manager that tracks connected clients), reach for `quiv.run_on_main`. Import it at module level and call it — `quiv` looks up the active instance, finds its main loop, and dispatches your callable. No `_progress_hook` parameter threaded through every layer, no `run_coroutine_threadsafe` glue.
```python
from quiv import run_on_main
async def broadcast(payload: dict):
await ws_manager.broadcast(payload) # lives on uvicorn's loop
def deeply_nested_step():
# Inside a Quiv task — but the broadcast hops to the main loop.
run_on_main(broadcast, {"event": "step_done"})
```
The same `run_on_main` call also works when invoked from a FastAPI route handler on the main loop — it detects that the caller is already on the main loop and schedules the work there directly (sync targets run inline, async targets are scheduled with `create_task`) instead of doing a cross-thread hop. One helper, two contexts.
### Correlate logs for one job, across threads
Every invocation gets a `_job_id` (UUID). Stamp it into a `LoggerAdapter` (or a `ContextVar`) and every log line from that run carries the same trace id — filtering logs by a single job is one query, even when N tasks run concurrently.
```python
import logging
base_logger = logging.getLogger(__name__)
def download_trailer(media_id: int, _job_id: str | None = None, _stop_event=None):
logger = logging.LoggerAdapter(base_logger, {"trace_id": _job_id})
logger.info("Starting download for media %s", media_id)
# every log line through `logger` below carries trace_id=<_job_id>
```
Trailarr uses a `ContextVar` flavor of this in production so downstream modules pick up the trace id automatically — see [Getting Started](getting-started.md) for that variant.
## Concepts
- **Task**: scheduling definition (`interval`, `run_once`, args/kwargs, status)
- **Job**: one execution record of a task
- **Task statuses**: `active`, `running`, `paused`
- **Job statuses**: `scheduled`, `running`, `completed`, `cancelled`, `failed`
## Important caveats
- **Temporary database**: each `Quiv` instance creates a temporary SQLite file that is deleted on `shutdown()`. Task/job state does not persist across restarts.
- **Single-process**: the scheduler runs in-process. It is not designed for distributed or multi-process deployments.
- **Picklable args**: `args` and `kwargs` passed to `add_task()` are pickle-serialized for persistence. Most Python objects are supported, but lambdas and inner functions are not picklable. The temporary SQLite database is trusted internal state — only your application code writes to it, and it is deleted on `shutdown()`. Do not expose the database file to untrusted input.
## Next pages
Interested in learning more or ready to start building with `quiv`? The full documentation is here:
- [Getting Started](getting-started.md) — install, scheduler setup, and your first task
- [API](api.md) — full reference for `Quiv`, `add_task`, and friends
- [Architecture](architecture.md) — how the scheduler, persistence, and execution layers fit together
- [Running on the main event loop](run-on-main.md) — dispatch work to the main loop from anywhere in a task's call stack
- [Event Listeners](event-listeners.md) — hook into task and job lifecycle events
- [Exceptions](exceptions.md) — the `QuivError` hierarchy and when each is raised
- [Testing](testing.md) — patterns for testing handlers and the scheduler in your suite
- [AI Tools](ai-tools.md) — bundled agent guide, llms.txt, and the Claude Code plugin
## Ideas, bugs, and contributions
`quiv` started from one app's needs, so the best way it gets better is when other people's apps push it in new directions. If you have a use case it doesn't cover, a rough edge it should smooth out, or a PR you'd like to land — all welcome.
- [Open an issue](https://github.com/nandyalu/quiv/issues) for bugs or feature requests
- [Start a discussion](https://github.com/nandyalu/quiv/discussions) if you'd like to talk through an idea first
- PRs are welcome — for anything non-trivial, opening an issue first is usually the fastest path
And if `quiv` saved you some time, a [GitHub star](https://github.com/nandyalu/quiv) is a nice way to let us know it was useful.
---
# Getting Started
This guide gets `quiv` running with recurring tasks, progress callbacks, and clean shutdown behavior.
## Install
=== "uv"
```bash
uv add quiv
```
=== "pip"
```bash
pip install quiv
```
For local development:
=== "uv"
```bash
git clone https://github.com/nandyalu/quiv.git
cd quiv
uv pip install -e ".[dev]"
```
=== "pip"
```bash
git clone https://github.com/nandyalu/quiv.git
cd quiv
pip install -e ".[dev]"
```
## 1) Create a scheduler
You can configure `Quiv` with either a `QuivConfig` object or direct args.
```python
from quiv import Quiv, QuivConfig
scheduler = Quiv(
config=QuivConfig(
pool_size=8, # default is 10
history_retention_seconds=3600, # default is 86400 (1 day)
timezone="UTC", # default is UTC
)
)
```
Equivalent direct parameters:
```python
from quiv import Quiv
scheduler = Quiv(
pool_size=8, # default is 10
history_retention_seconds=3600, # default is 86400 (1 day)
timezone="UTC", # default is UTC
)
```
Do not mix `config=...` with direct constructor config args. See [Quiv API](./api.md#quiv) for full configuration options.
## 2) Add a task
### Sync handler
```python
def my_task(
_job_id: str | None = None,
_stop_event: threading.Event | None = None,
_progress_hook: Callable | None = None,
):
total = 5
for step in range(1, total + 1):
#
if _progress_hook:
_progress_hook(step=step, total=total)
if _stop_event and _stop_event.is_set():
return
task_id = scheduler.add_task(
task_name="demo-task",
func=my_task,
interval=10,
delay=0,
run_once=False,
args=(),
kwargs={},
)
```
### Async handler
Async handlers are fully supported. They run in thread-local event loops created per invocation, so they do not block the scheduler or main loop.
```python
import httpx
async def poll_api(
_stop_event: threading.Event | None = None,
_progress_hook: Callable | None = None,
):
async with httpx.AsyncClient() as client:
# example of doing some async work
response = await client.get("https://api.example.com/status")
if _progress_hook:
_progress_hook(status_code=response.status_code)
if _stop_event and _stop_event.is_set():
return
scheduler.add_task(
task_name="api-poll",
func=poll_api,
interval=30,
)
```
`_job_id`, `_stop_event`, and `_progress_hook` are injected only if your handler accepts those keyword parameters. If your handler signature does not include them (and does not use `**kwargs`), they are not injected. See [Progress Callbacks](progress-callbacks.md) and [Cancellation](cancellation.md) for in-depth guides.
!!! tip "Hold onto `task_id`"
`add_task()` returns a `task_id` (UUID string). All runtime operations — `pause_task()`, `resume_task()`, `run_task_immediately()`, `remove_task()`, and `get_task()` — use this id. Multiple tasks can share the same `task_name`; each gets its own unique `task_id`.
## 3) Add progress callback (optional)
Progress callbacks can be sync or async. When an asyncio event loop is available, async callbacks run via `run_coroutine_threadsafe` and sync callbacks run via `call_soon_threadsafe` on the main loop. If no event loop is available (e.g. in a plain script without asyncio), sync callbacks run directly on the worker thread and async callbacks run in a temporary event loop on the worker thread.
```python
async def on_progress(**payload):
print("progress", payload)
scheduler.add_task(
task_name="demo-task-with-progress",
func=my_task,
interval=10,
progress_callback=on_progress,
)
```
## 4) Listen for events (optional)
Event listeners let you react to task and job lifecycle events. Register a callback with `add_listener()`:
```python
from quiv import Event
from quiv.models import Task, Job
def on_job_completed(event: Event, task: Task, job: Job):
print(f"Job {job.id} for '{task.task_name}' completed in {job.duration_seconds}s")
def on_job_failed(event: Event, task: Task, job: Job):
print(f"Job {job.id} for '{task.task_name}' failed: {job.error_message}")
scheduler.add_listener(Event.JOB_COMPLETED, on_job_completed)
scheduler.add_listener(Event.JOB_FAILED, on_job_failed)
```
!!! info "Typed callbacks"
`TASK_*` listeners receive `(event, task)`. `JOB_*` listeners receive `(event, task, job)`. Both use typed model objects with full IDE autocomplete — no dict key lookups.
Listeners follow the same dispatch model as progress callbacks: async listeners run on the main loop, sync listeners run via `call_soon_threadsafe` (or directly on the calling thread when no loop is available). Exceptions in listeners are logged and swallowed. See [Event Listeners](event-listeners.md) for the full event list and dispatch details.
## 5) Start and stop
```python
import asyncio
async def main() -> None:
scheduler.startup()
await asyncio.sleep(25)
scheduler.shutdown()
asyncio.run(main())
```
Always call `shutdown()` (or `stop()`) when your app exits.
`startup()` / `shutdown()` is the recommended pair, but `start()` / `stop()` works identically — they are aliases.
## 6) Operate tasks at runtime
```python
task_id = scheduler.add_task(
task_name="demo-task",
func=my_task,
interval=10,
)
scheduler.run_task_immediately(task_id)
scheduler.pause_task(task_id)
scheduler.resume_task(task_id)
```
## 7) Cancel a running job
```python
jobs = scheduler.get_all_jobs(status="running")
for job in jobs:
scheduler.cancel_job(job.id)
```
Cancellation is cooperative: it sets the job's stop event. The handler must check `_stop_event.is_set()` to actually stop.
## 8) Inspect state
```python
tasks = scheduler.get_all_tasks(include_run_once=True)
jobs = scheduler.get_all_jobs()
failed_jobs = scheduler.get_all_jobs(status="failed")
```
## FastAPI integration example
`quiv` is intended for app-integrated task scheduling, especially in FastAPI. Use the `lifespan` context manager to tie scheduler lifecycle to the app:
```python
from contextlib import asynccontextmanager
from fastapi import FastAPI
from quiv import Quiv
scheduler = Quiv(timezone="UTC")
def reindex_documents(_stop_event=None, _progress_hook=None) -> None:
total = 100
for step in range(1, total + 1):
if _stop_event and _stop_event.is_set():
return
# Simulate blocking work
import time
time.sleep(0.05)
if _progress_hook:
_progress_hook(step=step, total=total, stage="reindex")
async def on_reindex_progress(**payload) -> None:
# Replace with websocket broadcast, logging, metrics, etc.
print("progress", payload)
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup
scheduler.add_task(
task_name="reindex-docs",
func=reindex_documents,
interval=300,
progress_callback=on_reindex_progress,
)
scheduler.start()
yield
# Shutdown
scheduler.shutdown()
app = FastAPI(lifespan=lifespan)
```
Why this matters:
- `_stop_event` makes long tasks cancel safely on shutdown.
- `_progress_hook` sends task progress back into FastAPI's async context.
- Scheduler lifecycle is tied cleanly to app lifecycle.
## Logging
`quiv` uses Python's standard `logging` module. If you do not configure logging, no output is produced (Python's default `NullHandler` behavior).
To see scheduler logs, configure the `"Quiv"` logger:
```python
import logging
logging.basicConfig(level=logging.INFO)
```
Or configure the `"Quiv"` logger directly for more control:
```python
import logging
quiv_logger = logging.getLogger("Quiv")
quiv_logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(asctime)s %(name)s %(levelname)s %(message)s"))
quiv_logger.addHandler(handler)
```
You can also inject your own logger instance:
```python
import logging
my_logger = logging.getLogger("myapp.scheduler")
scheduler = Quiv(logger=my_logger)
```
The library logs at these levels:
| Level | What is logged |
|---------|--------------------------------------------------------------|
| DEBUG | Database table creation, datetime normalization |
| INFO | Task added, scheduler loop start, job start/completion, cleanup |
| WARNING | Progress callback skipped (no event loop or main loop closed) |
| ERROR | Job failures, scheduler loop errors, progress callback errors |
A separate `"quiv.models"` logger emits DEBUG-level messages for datetime normalization. This logger is not configurable via the constructor and follows standard Python logging configuration.
## Troubleshooting
- **`ConfigurationError` on startup**: check `pool_size > 0` and `history_retention_seconds >= 0`.
- **`InvalidTimezoneError`**: use a valid IANA timezone name (for example `UTC` or `America/New_York`).
- **`HandlerNotRegisteredError` for immediate run**: call `add_task(...)` first, and use the returned `task_id`.
- **`TaskNotScheduledError`**: the task handler is registered, but the scheduled task row no longer exists in the database.
- **No log output**: configure Python logging (see [Logging](#logging) above).
- **Args/kwargs errors**: `args` and `kwargs` are pickle-serialized, so most Python objects are supported. If you encounter errors, ensure the objects are picklable (e.g. lambdas and inner functions are not).
---
# Bigger Applications
In larger FastAPI projects, code is split across multiple packages and modules. This guide shows how to structure `quiv` in that setup — shared scheduler instance, tasks defined in separate files, API endpoints for runtime control, WebSocket progress updates, and graceful cancellation.
## Project structure
```
myapp/
├── main.py # FastAPI app, lifespan, WebSocket
├── scheduler.py # Quiv instance (shared singleton)
├── tasks/
│ ├── __init__.py
│ ├── cleanup.py # DB cleanup task
│ └── report.py # Report generation task
└── routes/
├── __init__.py
└── tasks.py # Task control endpoints
```
## 1) Create the scheduler instance
Define the `Quiv` instance in its own module so every other file can import it. Do **not** call `start()` here — that happens in the FastAPI lifespan.
```python
# myapp/scheduler.py
from quiv import Quiv
scheduler = Quiv(
pool_size=4,
history_retention_seconds=7200,
timezone="America/New_York",
)
```
Since `Quiv` lazily resolves the asyncio event loop, this works at module level before FastAPI or uvicorn creates a loop.
## 2) Create a logging context (optional)
Create a logging context that holds the `trace_id` which can be later used for logs
```python
# myapp/config/logging_context.py
import asyncio
from contextvars import ContextVar, Token
from functools import wraps
import uuid
# Define a ContextVar to hold the Trace ID
trace_id_var: ContextVar[str | None] = ContextVar("trace_id", default=None)
def get_trace_id() -> str | None:
"""Returns the current Trace ID."""
return trace_id_var.get()
def get_new_trace_id() -> str:
"""Generates and returns a new Trace ID without setting it in the context."""
return str(uuid.uuid4())
def generate_trace_id(trace_id: str | None = None) -> Token:
"""Generates a new Trace ID and sets it in the context."""
if trace_id is None:
trace_id = str(uuid.uuid4())
return trace_id_var.set(trace_id)
def clear_trace_id(token: Token) -> None:
"""Clears the Trace ID from the context."""
trace_id_var.reset(token)
def with_logging_context(func):
"""Decorator to add a Trace ID to the logging context for the duration of the function call.
Args:
func: The function to decorate.
Returns:
The decorated function with Trace ID management.
"""
@wraps(func)
async def async_wrapper(*args, **kwargs):
# 1. Extract or Create
trace_id = kwargs.get("_job_id") or kwargs.get("trace_id")
token = generate_trace_id(trace_id) # Falls back to uuid4() internally
try:
return await func(*args, **kwargs)
finally:
clear_trace_id(token)
@wraps(func)
def sync_wrapper(*args, **kwargs):
# 1. Extract or Create
trace_id = kwargs.get("_job_id") or kwargs.get("trace_id")
token = generate_trace_id(trace_id)
try:
return func(*args, **kwargs)
finally:
clear_trace_id(token)
# Return the appropriate wrapper based on the source function
return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper
```
Make the logging adapter or logging handler retrieve the trace_id and add it to logs.
For full working code with logging context, stop events, and progress hooks, see [Trailarr](https://github.com/nandyalu/trailarr/).
## 3) Define tasks in separate files
Each task file imports the shared scheduler to register its task via `add_task()`. Tasks are plain functions — sync or async.
### Cleanup task (sync, with stop event)
```python
# myapp/tasks/cleanup.py
import logging
import threading
import time
from config.logging_context import with_logging_context
logger = logging.getLogger(__name__)
@with_logging_context
def cleanup_stale_records(
days: int,
_job_id: str | None = None,
_stop_event: threading.Event | None = None,
):
"""Delete records older than `days` from the database."""
# quiv injects _job_id and with_logging_context decorator stores it as trace_id
# logs handler will get it using get_trace_id and adds it to all logs
# so logs from the task will be logged with that trace_id
# Attach job_id as trace context for this run
batches = 10
for batch in range(1, batches + 1):
if _stop_event and _stop_event.is_set():
logger.info("Cleanup cancelled at batch %d/%d", batch, batches)
return
# ... delete a batch of old records ...
time.sleep(1) # simulate work
logger.info("Cleanup finished: processed %d batches", batches)
```
### Report task (sync, with stop event and progress hook)
```python
# myapp/tasks/report.py
import logging
import threading
import time
from typing import Callable
from config.logging_context import with_logging_context
logger = logging.getLogger(__name__)
@with_logging_context
def generate_report(
report_type: str,
_job_id: str | None = None,
_stop_event: threading.Event | None = None,
_progress_hook: Callable | None = None,
):
"""Generate a report with progress updates."""
steps = 5
for step in range(1, steps + 1):
if _stop_event and _stop_event.is_set():
logger.info("Report generation cancelled at step %d/%d", step, steps)
return
# ... do a chunk of report work ...
time.sleep(2) # simulate work
if _progress_hook:
_progress_hook(
step=step,
total=steps,
report_type=report_type,
)
logger.info("Report '%s' generated successfully", report_type)
```
## 4) Register tasks and wire up the lifespan
The main module registers tasks, starts the scheduler on startup, and shuts it down on teardown. This is also where you set up WebSocket-based progress callbacks.
```python
# myapp/main.py
import asyncio
import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from quiv import Event
from quiv.models import Task, Job
from myapp.scheduler import scheduler
from myapp.tasks.cleanup import cleanup_stale_records
from myapp.tasks.report import generate_report
logger = logging.getLogger(__name__)
# ---------- WebSocket connection manager ----------
class ConnectionManager:
"""Track active WebSocket connections for progress broadcasts."""
def __init__(self):
self.connections: list[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.connections.remove(websocket)
async def broadcast(self, message: dict):
for ws in self.connections:
try:
await ws.send_json(message)
except Exception:
pass
ws_manager = ConnectionManager()
# ---------- Progress callback ----------
async def on_report_progress(**payload):
"""Forward task progress to all connected WebSocket clients."""
logger.info("Report progress: %s", payload)
await ws_manager.broadcast({"event": "progress", "data": payload})
# ---------- Lifespan ----------
async def on_job_event(event: Event, task: Task, job: Job) -> None:
"""Forward job lifecycle events to WebSocket clients."""
payload: dict = {"event": event.value, "task": task.task_name, "job_id": job.id}
if job.duration_seconds is not None:
payload["duration_seconds"] = job.duration_seconds
if job.error_message is not None:
payload["error"] = job.error_message
await ws_manager.broadcast(payload)
@asynccontextmanager
async def lifespan(app: FastAPI):
# Register event listeners
scheduler.add_listener(Event.JOB_STARTED, on_job_event)
scheduler.add_listener(Event.JOB_COMPLETED, on_job_event)
scheduler.add_listener(Event.JOB_FAILED, on_job_event)
# Register and start tasks
scheduler.add_task(
task_name="db-cleanup",
func=cleanup_stale_records,
interval=3600,
kwargs={"days": 30},
)
scheduler.add_task(
task_name="weekly-report",
func=generate_report,
interval=604800,
delay=10,
kwargs={"report_type": "weekly-summary"},
progress_callback=on_report_progress,
)
scheduler.start()
yield
scheduler.shutdown()
app = FastAPI(lifespan=lifespan)
# ---------- WebSocket endpoint ----------
@app.websocket("/ws/progress")
async def progress_websocket(websocket: WebSocket):
await ws_manager.connect(websocket)
try:
while True:
await websocket.receive_text()
except WebSocketDisconnect:
ws_manager.disconnect(websocket)
```
## 5) API endpoints for runtime control
A separate router imports the same scheduler instance to expose task management endpoints.
```python
# myapp/routes/tasks.py
from fastapi import APIRouter, HTTPException
from myapp.scheduler import scheduler
router = APIRouter(prefix="/tasks", tags=["tasks"])
@router.post("/{task_id}/run")
def run_task_now(task_id: str):
"""Trigger a scheduled task to run immediately."""
try:
count = scheduler.run_task_immediately(task_id)
except Exception as e:
raise HTTPException(status_code=404, detail=str(e))
return {"queued": count}
@router.post("/{task_id}/pause")
def pause_task(task_id: str):
"""Pause a task by id."""
try:
scheduler.pause_task(task_id)
except Exception as e:
raise HTTPException(status_code=404, detail=str(e))
return {"status": "paused"}
@router.post("/{task_id}/resume")
def resume_task(task_id: str, delay: int = 0):
"""Resume a paused task, optionally with a delay."""
try:
scheduler.resume_task(task_id, delay=delay)
except Exception as e:
raise HTTPException(status_code=404, detail=str(e))
return {"status": "resumed"}
@router.get("/{task_id}")
def get_task(task_id: str):
"""Get a single task by id."""
try:
return scheduler.get_task(task_id)
except Exception as e:
raise HTTPException(status_code=404, detail=str(e))
@router.get("/")
def list_tasks():
"""List all scheduled tasks."""
return scheduler.get_all_tasks()
@router.get("/jobs")
def list_jobs(status: str | None = None):
"""List jobs, optionally filtered by status."""
return scheduler.get_all_jobs(status=status)
@router.post("/jobs/{job_id}/cancel")
def cancel_job(job_id: str):
"""Cancel a running job."""
cancelled = scheduler.cancel_job(job_id)
if not cancelled:
raise HTTPException(status_code=404, detail="Job not found or not running")
return {"status": "cancelled"}
```
!!! tip "Task IDs in your API"
Since `add_task()` returns a `task_id` (UUID string), you can store it in your application state or return it to clients. All runtime operations (`pause_task`, `resume_task`, `run_task_immediately`, `remove_task`) use `task_id` as the identifier.
`Task` and `Job` are SQLModel objects, so FastAPI serializes them directly — no manual conversion needed. All datetime fields (`next_run_at`, `started_at`, `ended_at`) are guaranteed to be timezone-aware UTC, so the JSON output will include a `+00:00` suffix that browsers can parse and display in the user's local timezone.
Register the router in your app:
```python
# add to myapp/main.py
from myapp.routes.tasks import router as tasks_router
app.include_router(tasks_router)
```
## Run the full example
A complete runnable version of this app is in the [`examples/fastapi_app`](https://github.com/nandyalu/quiv/tree/main/examples/fastapi_app) directory. From the repository root:
```bash
uv run uvicorn examples.fastapi_app.main:app --reload
```
Then open [http://127.0.0.1:8000/docs](http://127.0.0.1:8000/docs) for interactive API docs, or connect to `ws://127.0.0.1:8000/ws/progress` for live progress updates.
## Key takeaways
- **Single instance, shared everywhere.** Create `Quiv` in one module and import it wherever needed. This avoids multiple schedulers and duplicate DB files.
- **Module-level init is safe.** `Quiv()` does not require a running asyncio loop at creation time. The event loop is resolved lazily when progress callbacks fire.
- **Lifespan owns the lifecycle.** Call `start()` and `shutdown()` in the FastAPI lifespan so the scheduler is tied to the app process.
- **Tasks are plain functions.** Define them anywhere. They only need `_stop_event` and `_progress_hook` in their signature if they want cancellation or progress support. See [Cancellation](cancellation.md) and [Progress Callbacks](progress-callbacks.md) for detailed guides.
- **Progress goes through WebSocket.** Async progress callbacks run on FastAPI's event loop, so they can broadcast to WebSocket clients directly. See [Progress Callbacks](progress-callbacks.md) for dispatch details.
- **Event listeners for observability.** Use `add_listener()` to react to task and job lifecycle events. Async listeners run on the main loop, so they can broadcast to WebSocket clients alongside progress callbacks. See [Event Listeners](event-listeners.md) for the full event list.
---
# API
## `Quiv`
### Constructor
```python
Quiv(
config: QuivConfig | None = None,
pool_size: int = 10,
history_retention_seconds: int = 86400,
timezone: str | tzinfo = "UTC",
*,
logger: logging.Logger | logging.LoggerAdapter | None = None,
main_loop: asyncio.AbstractEventLoop | None = None,
)
```
If `config` is provided, do not also pass explicit `pool_size/history_retention_seconds/timezone`.
Parameters:
- `config`: grouped configuration object (see [`QuivConfig`](#quivconfig))
- `pool_size`: maximum number of tasks that can run concurrently (default 10). See [Choosing a pool size](#choosing-a-pool-size) below
- `history_retention_seconds`: how long finished job records are kept (default 86400 = 24 hours)
- `timezone`: [IANA timezone string](https://en.wikipedia.org/wiki/List_of_tz_database_time_zones) or `tzinfo` for display formatting (default `"UTC"`)
!!! note "Timezone is for display only"
`timezone` is only used to format datetime values in quiv's log output. All internal datetime handling (scheduling, persistence, job lifecycle) uses UTC regardless of this setting.
- `logger`: optional custom logger or `LoggerAdapter` instance; if not provided, a logger named `"Quiv"` is used. The library does not set a log level — configure it in your application (see [Logging](getting-started.md#logging))
!!! note "Logger scope"
The `logger` is only used for quiv's own internal logs (scheduler loop events, job lifecycle, cleanup, warnings, etc.). Task handler logs are **not** routed through this logger — use your own loggers inside your task handlers as usual.
- `main_loop`: optional asyncio event loop for progress callbacks and event listeners.
- If not provided, resolution is attempted at `start()` time via `asyncio.get_running_loop()`, and lazily on first callback dispatch if still unset.
- This means `Quiv()` can be instantiated at module level before any event loop is running (common in FastAPI apps).
- If no event loop is available when a callback fires, async callbacks run in a temporary event loop on the worker thread, and sync callbacks run directly.
### `add_task(...)`
```python
add_task(
task_name: str,
func: Callable[..., Any],
interval: float,
delay: float = 0,
run_once: bool = False,
fixed_interval: bool = True,
args: tuple | None = None,
kwargs: dict | None = None,
progress_callback: Callable[..., Any] | None = None,
) -> str
```
Adds a scheduled task and returns its task ID (UUID string)[^1].
[^1]: Hold onto the returned `task_id` — it is the key for all subsequent operations:
- `remove_task()`
- `pause_task()`
- `resume_task()`
- `run_task_immediately()`
- `get_task()`.
This is the primary way to register tasks. It handles handler registration, progress callback registration, and task persistence in one call.
Validation:
- `task_name` must not be empty
- `interval > 0`
- `delay >= 0`
!!! note "Duplicate task names are allowed"
Multiple tasks can share the same `task_name`. Each call to `add_task()` returns a unique `task_id` (UUID) which is the identifier used for all task operations. `task_name` is a display label, not a unique key.
Behavior:
- `func` may be sync or async
- `args`/`kwargs` are pickle-serialized and persisted — most Python objects are supported, but lambdas and inner functions are not picklable.
!!! warning
The temporary database is trusted internal state and should not be exposed to untrusted input. Doing so might open to attackers injecting untrusted args/kwargs that gets passed to Tasks which could compromise the application.
- if `run_once=True`, task is executed once and then removed from storage
- if `progress_callback` is provided, it runs on the main loop when available, or directly on the worker thread otherwise
!!! info "`fixed_interval` scheduling modes"
**`fixed_interval=True`** (default) — Next run is scheduled at fixed intervals from the job **start time**. A task with `interval=3600` runs every 3600 seconds relative to that start-time anchor (for example, if a run starts at 12:00:10, subsequent targets are 13:00:10, 14:00:10, etc.), regardless of how long the task takes. If a run exceeds the interval, missed intervals are skipped and the next run lands on the next scheduled time in that cadence.
**`fixed_interval=False`** — Next run is scheduled `interval` seconds after job **completion**. A task with `interval=3600` that takes 10 minutes to run will have 70-minute gaps between start times.
### `start() -> None` / `startup() -> None`
Starts scheduler background loop thread. Safe to call multiple times.
!!! success "`startup()` is an alias for `start()`"
use whichever reads better in your code. `startup()` pairs naturally with `shutdown()`.
### `shutdown(timeout: float | None = None) -> None` / `stop(...) -> None`
- Stops scheduler loop and worker threads
- Cancels running jobs via stop events
- Disposes DB engine
- And removes temporary scheduler SQLite file.
Always call this during app teardown.
With `timeout=None` (default) shutdown waits indefinitely for in-flight jobs to finish. Pass a `timeout` (seconds) to bound the wait: jobs that do not exit within the deadline are abandoned on their worker threads with a warning — useful in FastAPI lifespan teardown where a hung handler must not block application shutdown. Abandoned jobs may log errors afterwards (e.g. writing to the already-deleted database).
!!! success "`stop()` is an alias for `shutdown()`"
use whichever reads better in your code. `stop()` pairs naturally with `start()`.
### `run_task_immediately(task_id: str) -> int`
Queues an already-scheduled task to run now.
Raises:
- `HandlerNotRegisteredError` if no registered handler exists for the id
- `TaskNotScheduledError` if handler exists but no scheduled task row exists
- `TaskNotActiveError` if the task is `running` (no concurrent second run) or `paused` (use `resume_task()` instead)
Returns number of task rows queued.
### `pause_task(task_id: str) -> None`
Pause blocks future runs of the task.
Raises:
- `TaskNotFoundError` if no task with that id exists.
### `resume_task(task_id: str, delay: int = 0) -> None`
Resume re-activates and sets next run with an optional `delay` (in seconds, default=0).
Raises:
- `TaskNotFoundError` if no task with that id exists.
!!! info
If a `delay` is not provided or set to 0, next run will fire immediately.
### `cancel_job(job_id: str) -> bool`
Signals cancellation for a running job by setting its stop event.
Returns `True` if the stop event was found and set, `False` otherwise.
!!! info
Cancellation is cooperative: the handler must check `_stop_event.is_set()` to actually stop.
### `get_task(task_id: str) -> Task`
Returns a single [`Task`](#task) by its UUID string.
Raises:
- `TaskNotFoundError` if no task with that ID exists.
### `get_job(job_id: str) -> Job`
Returns a single [`Job`](#job) by its UUID string.
Raises:
- `JobNotFoundError` if no job with that ID exists.
### `get_all_tasks(include_run_once: bool = False) -> list[Task]`
Returns persisted task rows as [`Task`](#task) objects.
- when `include_run_once=False`, run-once tasks are excluded
- when `include_run_once=True`, all persisted tasks are returned
### `get_all_jobs(status: str | None = None) -> list[Job]`
Returns persisted jobs, optionally filtered by status string (e.g. `"failed"`, `"running"`).
### `remove_task(task_id: str) -> None`
Removes a scheduled task, its registered handler, and progress callback. If the task has a running job, its stop event is set to signal cancellation.
Raises:
- `TaskNotFoundError` if no task with that id exists.
Any previously running job will finish on its own and clean up normally.
### `add_listener(event: Event, callback: Callable[..., Any]) -> None`
Register an event listener for a scheduler lifecycle event. Multiple listeners can be registered for the same event. Both sync and async callbacks are supported.
The callback signature depends on the event group:
- **`TASK_*` events**: `callback(event: Event, task: Task)`
- **`JOB_*` events**: `callback(event: Event, task: Task, job: Job)`
Listeners receive typed `Task` and `Job` model objects with full IDE autocomplete — no dict key lookups needed.
Raises:
- `ConfigurationError` if `event` is not an `Event` enum member or `callback` is not callable.
See [Event Listeners](event-listeners.md) for the full event list and dispatch details.
### `remove_listener(event: Event, callback: Callable[..., Any]) -> None`
Remove a previously registered event listener. If the callback is not found, the call is silently ignored.
## `run_on_main`
```python
from quiv import run_on_main
run_on_main(func: Callable[..., Any], *args: Any, **kwargs: Any) -> None
```
Module-level helper that dispatches `func` onto the active Quiv instance's main event loop. Designed to be called from anywhere in a task handler's call stack — without threading a callback parameter through intermediate functions — and from code already running on the main loop (e.g., FastAPI route handlers that share utilities with task code).
Resolution order for the active instance:
1. A `ContextVar` set by `_run_job` for the duration of a handler invocation (propagates through nested sync calls, the per-job async loop, and `asyncio.create_task`).
2. A process-level fallback registered by `Quiv.start()` and cleared by `Quiv.shutdown()`.
Behavior:
- Fire-and-forget; returns `None` immediately on cross-thread dispatch.
- Sync targets run inline when called from the main loop's thread, else via `call_soon_threadsafe`. Async targets are scheduled via `main_loop.create_task` on-loop, else `run_coroutine_threadsafe`.
- Exceptions raised by `func` are logged on the active Quiv's logger and swallowed.
Raises:
- `RuntimeError` if no active Quiv instance is registered, or if the active instance has no resolvable main event loop.
See [Running on the main event loop](run-on-main.md) for the full walkthrough, dispatch table, and caveats.
## Hooks and callback injection
When a task is dispatched, `quiv` inspects handler signatures:
- injects `_job_id` (`str`, UUID) only if accepted
- injects `_stop_event` (`threading.Event`) only if accepted
- injects `_progress_hook` (callable) only if accepted
If your handler does not define those parameters (and does not use `**kwargs`), no injection is performed.
Async handlers run in thread-local event loops created per invocation. They do not share the main application event loop.
## Models
### `Task`
Public API model returned by `get_task()` and `get_all_tasks()`. Use directly in FastAPI endpoints — no manual conversion needed.
```python
# Methods return Task directly
task_id = scheduler.add_task("my-task", handler, interval=60)
task = scheduler.get_task(task_id) # Task
tasks = scheduler.get_all_tasks() # list[Task]
# Use directly in FastAPI endpoints
@app.get("/tasks")
def list_tasks():
return scheduler.get_all_tasks()
```
Key fields:
- `id: str` — UUID identifier
- `task_name: str` — display name (not necessarily unique)
- `args: tuple[Any, ...]` — positional arguments (unpickled)
- `kwargs: dict[str, Any]` — keyword arguments (unpickled)
- `interval_seconds: float` — seconds between runs
- `fixed_interval: bool` — if `True`, next run is measured from job start time; if `False`, from completion
- `run_once: bool` — if `True`, task runs once then is removed
- `status: str` — `"active"`, `"running"`, or `"paused"`
- `next_run_at: datetime` — next scheduled run (UTC-aware)
!!! abstract "datetime objects are in UTC"
The datetime values (`next_run_at`) are always returned as a UTC-aware datetime.
- You can safely return this from fastapi endpoints which will have a `Z` at the end to indicate UTC datetime.
- This is the golden-standard for Browsers as they can easily parse it and display in user's timezone.
!!! info "Internal TaskDB model"
Internally, quiv stores `args` and `kwargs` as pickle-encoded bytes in the `TaskDB` model for flexibility. The public API automatically converts to `Task` with unpickled values and correct types for JSON/OpenAPI.
### `Job`
Key fields:
- `id: str` — UUID identifier
- `task_id: str` — foreign key to source task
- `task_name: str` — name of the task that spawned this job
- `status: str` — lifecycle status
- `started_at: datetime` — UTC-aware start timestamp
- `ended_at: datetime | None` — UTC-aware end timestamp
- `duration_seconds: float | None` — job duration in seconds (set on completion)
- `error_message: str | None` — error description if job failed
!!! abstract "datetime objects are in UTC"
The datetime values (`started_at`, `ended_at`) are always returned as UTC-aware datetimes.
- You can safely return this from fastapi endpoints which will have a `Z` at the end to indicate UTC datetime.
- This is the golden-standard for Browsers as they can easily parse it and display in user's timezone.
## Event types
### `Event`
- `task_added` — fired after a task is registered
- `task_removed` — fired after a task is removed
- `task_paused` — fired after a task is paused
- `task_resumed` — fired after a task is resumed
- `job_started` — fired when a job begins execution
- `job_completed` — fired when a job finishes successfully
- `job_failed` — fired when a job ends with an exception
- `job_cancelled` — fired when a job is cancelled
See [Event Listeners](event-listeners.md) for the data each event carries.
## Status constants
### `TaskStatus`
- `active` — task is eligible for scheduling
- `running` — task is currently executing
- `paused` — task is temporarily disabled
### `JobStatus`
- `scheduled` — job is queued for execution
- `running` — job is currently executing
- `completed` — job finished successfully
- `cancelled` — job stopped via cancellation signal
- `failed` — job ended with an exception
## `QuivConfig`
```python
QuivConfig(
pool_size: int = 10,
history_retention_seconds: int = 86400,
timezone: str | tzinfo = "UTC",
)
```
Frozen dataclass. Both `QuivConfig` and `Quiv` use `timezone` for the display timezone parameter.
## Choosing a pool size
`pool_size` controls the maximum number of tasks that can run concurrently. It is **not** tied to CPU cores — quiv uses threads, not processes, so the deciding factor is your workload, not hardware.
**What to consider:**
- **How many tasks might overlap?** If you have 5 recurring tasks and at most 3 could run at the same time, `pool_size=4` is sufficient.
- **Are tasks I/O-bound or CPU-bound?** I/O-bound tasks (API calls, database queries, file downloads) spend most of their time waiting, so many threads work fine. CPU-bound tasks contend for Python's GIL — more threads won't help and can hurt. For CPU-heavy work, offload to a process pool from within the handler rather than increasing `pool_size`.
- **Do tasks hold external resources?** Database connections, API rate limits, and file handles create practical caps regardless of thread count.
**Rules of thumb:**
- Start with the default (`10`) and only adjust if you see the `threadpool was busy` warning in your logs.
- For mostly I/O-bound workloads, set `pool_size` to 2–3x your expected max concurrent tasks.
- If the warning appears frequently, increase `pool_size` or check whether tasks are taking longer than expected.
When the pool is full, quiv defers due tasks to the next scheduler tick rather than queuing them unboundedly. If a job starts late because all workers were busy, a warning is logged with the delay.
## Public methods summary
- `Quiv(...)` — create scheduler instance
- `run_on_main(func, *args, **kwargs)` — fire-and-forget dispatch onto the active Quiv's main loop
- `add_task(...)` — schedule a task, returns `task_id`
- `start()` / `startup()` — start the scheduler loop
- `shutdown(timeout=None)` / `stop(timeout=None)` — stop scheduler and clean up resources
- `run_task_immediately(task_id)` — trigger a scheduled task now
- `pause_task(task_id)` — pause a task
- `resume_task(task_id)` — resume a paused task
- `cancel_job(job_id)` — signal cancellation for a running job
- `remove_task(task_id)` — remove a task and its registrations
- `add_listener(event, callback)` — register an event listener
- `remove_listener(event, callback)` — remove an event listener
- `get_task(task_id)` — get a single task by UUID
- `get_job(job_id)` — get a single job by ID
- `get_all_tasks(...)` — list persisted tasks
- `get_all_jobs(...)` — list persisted jobs
---
# Architecture
`quiv` is split into focused layers:
- `base` layer (`quiv/base.py`): runtime lifecycle, DB bootstrap, threadpool, callback plumbing, cancellation controls
- `scheduler` layer (`quiv/scheduler.py`): public API and scheduling loop
- `persistence` layer (`quiv/persistence.py`): task/job storage operations
- `execution` layer (`quiv/execution.py`): invocation preparation and sync/async dispatch
- `models` layer (`quiv/models.py`): SQLModel entities and status constants
## Runtime flow
```mermaid
sequenceDiagram
participant App as Application
participant Q as Quiv
participant DB as SQLite
participant Pool as ThreadPool
participant H as Handler
App->>Q: Quiv() — init
Q->>DB: Create temp DB + tables
App->>Q: add_listener(event, callback)
Note over Q: Register in _event_listeners dict
App->>Q: add_task()
Q->>DB: INSERT Task row
Q->>App: Emit TASK_ADDED event
App->>Q: start()
Note over Q: Scheduler loop thread starts
loop Every 1 second
Q->>Q: Check backpressure
Q->>DB: SELECT due active tasks
DB-->>Q: Due tasks
Q->>DB: Mark task as RUNNING
Q->>DB: INSERT Job row
Q->>Pool: Submit job
Pool->>H: Execute handler
Note over H: _job_id / _stop_event / _progress_hook injected if accepted
Pool->>App: Emit JOB_STARTED event
H-->>Pool: Return result
Pool->>App: Emit JOB_COMPLETED/FAILED event
Pool->>DB: Finalize job status
Pool->>DB: Set task ACTIVE, schedule next_run
end
App->>Q: shutdown()
Q->>Q: Cancel tracked jobs
Q->>Pool: Shutdown executor
Q->>DB: Dispose engine + delete DB files
```
1. `Quiv(...)` initializes runtime resources
- resolves timezone
- creates temporary SQLite database in OS temp directory
- initializes SQLModel tables
- creates threadpool executor
2. `add_listener(event, callback)` registers a callback for scheduler events.
- Listeners can be added at any time.
- Multiple listeners per event are supported.
3. `add_task(...)` creates a `Task` row with scheduling metadata, then registers the handler and progress callback by `task_id`.
- Returns a unique `task_id` (UUID string) used for all subsequent task operations.
- Multiple tasks can share the same `task_name` — each gets its own `task_id`.
- Emits `TASK_ADDED` event.
- Tasks can be added before `start()`, after `start()`, or at any point while the scheduler is running.
4. `start()` launches scheduler loop thread.
5. Loop iteration (runs every 1 second):
- cleans old job history via SQL-level DELETE (every 60 seconds, not every tick)
- checks backpressure: skips dispatch if all workers are busy
- selects due active tasks (`next_run_at <= now`, `status == active`)
- marks task as `running` — prevents concurrent runs of the same task
- creates a `Job` row for each due task
- prepares invocation args (inject hooks if supported)
- submits execution to threadpool
- emits `JOB_STARTED` event
6. Job completion:
- emits `JOB_COMPLETED`, `JOB_FAILED`, or `JOB_CANCELLED` event
- updates job with terminal status (`completed`, `failed`, `cancelled`)
- sets task back to `active` and schedules next run (from start time when `fixed_interval=True`, from completion when `False`)
- for run-once tasks, deletes the task row instead
- jobs that started late due to pool saturation log a warning with the delay
## Cancellation model
- each job receives its own `threading.Event` stop signal if handler accepts `_stop_event`
- `cancel_job(job_id)` sets that event when the job is currently tracked
- cancellation is cooperative: handler code must check the event
For writing cancellable handlers, shutdown behavior, and status determination logic, see [Cancellation](cancellation.md).
## Progress callback model
- handlers can receive `_progress_hook` when accepted in signature
- calling `_progress_hook(...)` dispatches configured progress callback via `_resolve_main_loop()`
- the main event loop is lazily resolved on first dispatch — `Quiv()` can be instantiated at module level before any asyncio loop exists
- with an event loop available:
- async callbacks are dispatched via `run_coroutine_threadsafe`
- sync callbacks are dispatched via `call_soon_threadsafe`
- without an event loop (e.g. plain scripts without asyncio):
- sync callbacks run directly on the worker thread
- async callbacks run in a temporary event loop on the worker thread
For dispatch flow details, async/sync examples, and error handling, see [Progress Callbacks](progress-callbacks.md).
## Event listener model
- listeners are registered globally via `add_listener(event, callback)`
- multiple listeners per event are supported, called in registration order
- dispatch uses the same mechanism as progress callbacks:
- async listeners dispatched via `run_coroutine_threadsafe` on the main loop
- sync listeners dispatched via `call_soon_threadsafe` on the main loop
- without a loop: async listeners run in a temporary event loop, sync listeners run directly
- listener exceptions are logged and swallowed — they never block the scheduler or fail a job
- task events (`TASK_ADDED`, `TASK_REMOVED`, `TASK_PAUSED`, `TASK_RESUMED`) fire on the calling thread (whoever called `add_task()`, etc.)
- job events (`JOB_STARTED`, `JOB_COMPLETED`, `JOB_FAILED`, `JOB_CANCELLED`) fire from the worker thread executing the job
For event types, data payloads, and examples, see [Event Listeners](event-listeners.md).
## Async execution model
Async task handlers do not run on the main application event loop. Instead, each async invocation creates a dedicated thread-local event loop, runs the coroutine to completion, and tears down the loop. This ensures async handlers do not interfere with each other or with the main loop.
## Persistence model
- tasks and jobs are persisted in internal SQLite tables:
- `quiv_task`
- `quiv_job`
- `quiv` uses a private SQLAlchemy `registry` to keep its metadata separate from user-defined SQLModel models
- datetimes are normalized to UTC-aware values on model load
- history cleanup removes old finished jobs by retention cutoff
## Thread safety
- the scheduler loop runs in a single daemon thread
- task handlers execute in the threadpool (`ThreadPoolExecutor`)
- each handler invocation gets its own stop event and kwargs; there is no shared mutable state between concurrent handler runs
- persistence operations use short-lived `Session` scopes
- progress callbacks are dispatched thread-safely onto the main asyncio loop when available, or run directly on the worker thread when no loop exists
## Lifecycle and teardown
- `shutdown()`:
- requests loop shutdown
- signals cancellation for tracked running jobs
- joins scheduler thread
- shuts down threadpool
- disposes engine and removes temp DB file
- the temporary SQLite database does not survive process restarts
---
# Progress Callbacks
Progress callbacks let task handlers report live progress back to your application. This is useful for updating UIs, broadcasting WebSocket messages, logging metrics, or tracking long-running work.
## How it works
When a handler calls `_progress_hook(...)`, quiv dispatches the registered progress callback for that task. The dispatch path depends on whether an asyncio event loop is available and whether the callback is sync or async.
```mermaid
flowchart TD
A["Handler calls _progress_hook(...)"] --> B{Callback registered?}
B -- No --> C[Return silently]
B -- Yes --> D{Event loop available?}
D -- Yes --> E{Async callback?}
D -- No --> F{Async callback?}
E -- Yes --> G["run_coroutine_threadsafe()
on main loop"]
E -- No --> H["call_soon_threadsafe()
on main loop"]
F -- Yes --> I["Run in temporary event loop"]
F -- No --> J["Call directly on
worker thread"]
```
### The four dispatch paths
| Event loop | Callback type | What happens |
|------------|--------------|--------------|
| Available | Async | Dispatched via `run_coroutine_threadsafe` on the main loop |
| Available | Sync | Dispatched via `call_soon_threadsafe` on the main loop |
| Unavailable | Sync | Called directly on the worker thread |
| Unavailable | Async | Run in a temporary event loop on the worker thread |
## Event loop resolution
quiv does **not** require an event loop at startup. The main loop is lazily resolved the first time a progress callback fires:
```mermaid
sequenceDiagram
participant App as Application
participant Q as Quiv()
participant UV as Uvicorn / asyncio
App->>Q: Quiv() at module level
Note over Q: _main_loop = None
UV->>UV: Event loop starts
App->>Q: scheduler.start()
Note over Q: Loop thread begins
Q->>Q: Handler calls _progress_hook
Q->>Q: _resolve_main_loop()
Q->>UV: asyncio.get_running_loop()
Note over Q: _main_loop cached
Q->>UV: Dispatch callback on loop
```
This means `Quiv()` can be instantiated at module level before FastAPI or uvicorn creates an event loop — the common pattern for larger applications.
## Adding a progress callback
Pass a `progress_callback` when adding a task:
```python
async def on_progress(**payload):
print("progress", payload)
scheduler.add_task(
task_name="my-task",
func=my_handler,
interval=60,
progress_callback=on_progress,
)
```
## Writing a handler with progress reporting
Add `_progress_hook` to your handler's signature. quiv inspects the signature and only injects it if the parameter is present.
```python
import threading
from typing import Callable
def process_records(
batch_size: int,
_stop_event: threading.Event | None = None,
_progress_hook: Callable | None = None,
):
records = fetch_records(batch_size)
total = len(records)
for i, record in enumerate(records, 1):
if _stop_event and _stop_event.is_set():
return
process(record)
if _progress_hook:
_progress_hook(
step=i,
total=total,
pct=round(i / total * 100),
)
```
The handler does not need to know whether the callback is sync or async, or whether an event loop exists. It just calls `_progress_hook(...)` and quiv handles the dispatch.
## Async progress callback
Async callbacks run on the main event loop via `run_coroutine_threadsafe`. This is ideal for FastAPI apps where you want to broadcast to WebSocket clients:
```python
from fastapi import WebSocket
connected_clients: list[WebSocket] = []
async def on_progress(**payload):
for ws in connected_clients:
await ws.send_json({"event": "progress", "data": payload})
scheduler.add_task(
task_name="etl-pipeline",
func=run_etl,
interval=3600,
progress_callback=on_progress,
)
```
Since the callback runs on FastAPI's event loop, you can safely use `await` with WebSockets, database sessions, or any async API.
## Sync progress callback
Sync callbacks work identically from the handler's perspective. When an event loop is available, they run on the main loop via `call_soon_threadsafe`. When no loop is available (e.g. a plain script), they run directly on the worker thread.
```python
import logging
logger = logging.getLogger(__name__)
def log_progress(**payload):
logger.info("Task progress: %s", payload)
scheduler.add_task(
task_name="cleanup",
func=cleanup_handler,
interval=300,
progress_callback=log_progress,
)
```
## Without an event loop
In scripts that don't use asyncio, sync progress callbacks still work — they run directly on the worker thread that executes the handler:
```python
from quiv import Quiv
scheduler = Quiv()
def on_progress(**payload):
print(f"Step {payload['step']}/{payload['total']}")
def my_task(_progress_hook=None):
for i in range(1, 6):
if _progress_hook:
_progress_hook(step=i, total=5)
scheduler.add_task(
task_name="script-task",
func=my_task,
interval=10,
progress_callback=on_progress,
)
scheduler.start()
```
Async progress callbacks also work in this scenario — they run in a temporary event loop on the worker thread, so `await` calls inside the callback will execute correctly.
## Error handling
If a progress callback raises an exception, quiv logs the error but does **not** fail the job. The handler continues running. This prevents a broken callback from disrupting task execution.
```mermaid
flowchart TD
A[Handler runs] --> B["_progress_hook(...)"]
B --> C[Callback dispatched]
C --> D{Callback raises?}
D -- No --> E[Continue]
D -- Yes --> F[Log error]
F --> E
```
## Payload conventions
`_progress_hook` accepts any `*args` and `**kwargs`. There is no enforced schema, but a useful pattern is:
```python
_progress_hook(
step=3, # current step
total=10, # total steps
stage="load", # descriptive label
pct=30, # percentage complete
)
```
The progress callback receives exactly what the handler passes — quiv uses the `task_id` internally to look up the registered callback, but this is consumed by the dispatch layer and not forwarded to the callback.
---
# Running on the main event loop
`quiv.run_on_main` is a fire-and-forget helper that dispatches a callable onto the main event loop from anywhere in a task handler's call stack — no matter how deeply nested — and works identically when called from code already running on the main loop (e.g., a FastAPI route handler).
It is the right tool when a piece of code, sometimes shared between task handlers and request handlers, needs to hop onto the main loop to touch resources that live there (WebSocket managers, background queues, async clients) but you don't want to thread a callback parameter through every intermediate function.
## When to use this vs `_progress_hook`
| Use case | Reach for |
| ------------------------------------------------------------- | ------------------- |
| A single progress channel per task, registered up front | `_progress_hook` |
| Ad-hoc main-loop work from deeply nested code | `run_on_main` |
| The same utility called from both task code and route handlers | `run_on_main` |
`_progress_hook` is per-task and is dispatched only to the `progress_callback` you registered with `add_task()`. `run_on_main` is global to the active Quiv instance — any callable, any arguments, called from any depth in any task.
## Basic usage
```python
from quiv import Quiv, run_on_main
scheduler = Quiv()
async def broadcast(message: str) -> None:
# Runs on uvicorn's main loop where ws_manager lives.
await ws_manager.broadcast(message)
def level_three() -> None:
run_on_main(broadcast, "deep call finished")
def level_two() -> None:
level_three()
def level_one() -> None:
level_two()
def handler() -> None:
level_one()
scheduler.add_task(task_name="deep-task", func=handler, interval=60)
scheduler.start()
```
`level_three` does not need `_progress_hook` injected, does not need a scheduler reference, and does not need to know it is inside a task — it just imports `run_on_main` at module level and calls it.
## Dispatch behavior
`run_on_main` checks at call time whether it is already on the main loop's thread and chooses the right dispatch path automatically.
```mermaid
flowchart TD
A["run_on_main(func, *args, **kwargs)"] --> B{Active Quiv instance?}
B -- No --> R[Raise RuntimeError]
B -- Yes --> C{Main loop resolvable?}
C -- No --> R
C -- Yes --> D{On main loop's thread?}
D -- Yes --> E{Async target?}
D -- No --> F{Async target?}
E -- Yes --> G["main_loop.create_task(coro)"]
E -- No --> H["Call inline; log + swallow exceptions"]
F -- Yes --> I["run_coroutine_threadsafe(coro, main_loop)"]
F -- No --> J["call_soon_threadsafe(wrapped, main_loop)"]
```
| Caller is on... | Sync target | Async target |
| ------------------------- | ---------------------------------------- | --------------------------------------- |
| Main loop's thread | Inline call (with exception swallowing) | `main_loop.create_task(coro)` |
| Worker thread | `call_soon_threadsafe` | `run_coroutine_threadsafe` |
All paths are **fire-and-forget**: `run_on_main` returns `None` immediately on cross-thread dispatch, and the call to `func` is allowed to fail without affecting the caller — see [Exception handling](#exception-handling).
## Same function, task or route
Because `run_on_main` resolves the active Quiv instance lazily, you can use the same utility function from a task handler **or** from a FastAPI route on the main loop. When called from the main loop's thread, no cross-thread dispatch happens — the sync target runs inline and the async target is scheduled on the current loop.
```python
from quiv import run_on_main
async def notify_clients(payload: dict) -> None:
await ws_manager.broadcast(payload)
# Called from inside a Quiv task (worker thread):
def task_handler() -> None:
run_on_main(notify_clients, {"event": "task_done"})
# Called from inside a FastAPI route (main loop):
@app.post("/notify")
async def notify_endpoint(payload: dict) -> dict:
run_on_main(notify_clients, payload)
return {"queued": True}
```
## How the active Quiv instance is resolved
Resolution order at call time:
1. A `ContextVar` set by `Quiv._run_job` for the duration of a handler invocation. This is set on the worker thread before the handler runs and propagates into:
- nested sync function calls,
- async handlers running in the per-job event loop,
- `asyncio.create_task` spawned inside an async handler.
2. A process-level fallback registered by `Quiv.start()` and cleared by `Quiv.shutdown()`. This covers callers that are **not** inside a task context, such as FastAPI route handlers on the main loop.
If both are unset, `run_on_main` raises `RuntimeError`.
!!! info "Multiple Quiv instances"
If more than one Quiv instance is `start()`ed at the same time, a warning is logged and the most recently started instance wins for out-of-task callers. Inside a task, the contextvar always points at the instance that scheduled the running job, regardless of how many other instances exist.
!!! warning "User-spawned threads do not inherit context"
Python's `ContextVar` is copied into `asyncio.create_task` but **not** into manually-spawned `threading.Thread` objects. If your handler does `threading.Thread(target=fn).start()` and `fn` calls `run_on_main`, the helper falls back to the process-level instance — which is usually still correct, but will be ambiguous if you run multiple Quiv instances. Avoid spawning raw threads from inside a handler if you can; the threadpool already gives you parallelism.
## Exception handling
If `func` raises, `run_on_main` logs the error via the active Quiv's logger and swallows it. The caller's code continues normally. This mirrors `_progress_hook` and event-listener semantics — a broken main-loop callback should never take down the task that triggered it.
```mermaid
flowchart TD
A["run_on_main(func, ...)"] --> B[Dispatch to main loop]
B --> C{func raises?}
C -- No --> D[Done]
C -- Yes --> E[Log error on Quiv logger]
E --> D
```
If you need failures to propagate, do the error handling inside `func` itself — for example, by reporting back through a `progress_callback` or an event listener.
## Signature and semantics
```python
def run_on_main(
func: Callable[..., Any],
*args: Any,
**kwargs: Any,
) -> None: ...
```
- `func` may be a sync function or a coroutine function (`async def`). Bare coroutine objects are **not** accepted — pass the function and let `run_on_main` invoke it with `*args, **kwargs`.
- Returns `None`. The call returns immediately for all cross-thread dispatch paths and for async targets scheduled on the current loop; the one exception is a **sync** target called from the main loop's thread, which runs inline on the caller's stack and therefore blocks the caller for the duration of the target.
- Raises `RuntimeError` if no active Quiv instance is registered, or if the active Quiv has no resolvable main loop. These are configuration errors (e.g., `run_on_main` called before `Quiv.start()`).
## A note on blocking the main loop
When called from the main loop's thread with a **sync** target, the target runs inline on the current call stack — exactly as if you had called it directly. If the target does blocking I/O or heavy CPU work it will block the loop. This is not a regression introduced by `run_on_main`; it is how Python coroutines call sync code in general. Use async targets, or schedule the blocking work onto a thread pool yourself.
---
# Event Listeners
Event listeners let you react to scheduler lifecycle events — tasks being added, removed, paused, or resumed, and jobs starting, completing, failing, or being cancelled. This is useful for logging, metrics, alerting, or updating UI state.
## How it works
Register a callback for one or more `Event` types via `add_listener()`. When the event fires, quiv dispatches your callback on the main event loop (same dispatch model as [progress callbacks](progress-callbacks.md)).
```mermaid
flowchart TD
A["Scheduler emits event"] --> B{Listeners registered?}
B -- No --> C[Return silently]
B -- Yes --> D{Event loop available?}
D -- Yes --> E{Async listener?}
D -- No --> F{Async listener?}
E -- Yes --> G["run_coroutine_threadsafe()
on main loop"]
E -- No --> H["call_soon_threadsafe()
on main loop"]
F -- Yes --> I["Run in temporary event loop"]
F -- No --> J["Call directly on
calling thread"]
```
### Dispatch paths
| Event loop | Listener type | What happens |
|------------|--------------|--------------|
| Available | Async | Dispatched via `run_coroutine_threadsafe` on the main loop |
| Available | Sync | Dispatched via `call_soon_threadsafe` on the main loop |
| Unavailable | Sync | Called directly on the calling thread |
| Unavailable | Async | Run in a temporary event loop on the calling thread |
## Events
All events are defined in the `Event` enum:
| Event | When it fires | Callback receives |
|-------|--------------|-------------------|
| `TASK_ADDED` | After `add_task()` completes | `event`, `task` |
| `TASK_REMOVED` | After `remove_task()` completes | `event`, `task`[^1] |
| `TASK_PAUSED` | After `pause_task()` completes | `event`, `task` |
| `TASK_RESUMED` | After `resume_task()` completes | `event`, `task` |
| `JOB_STARTED` | When a job begins execution | `event`, `task`, `job` |
| `JOB_COMPLETED` | When a job finishes successfully | `event`, `task`, `job` |
| `JOB_FAILED` | When a job ends with an exception | `event`, `task`, `job` |
| `JOB_CANCELLED` | When a job is cancelled via stop event | `event`, `task`, `job` |
[^1]: For `TASK_REMOVED`, the `task` object is a snapshot taken before deletion.
## Callback signatures
Listeners receive typed model objects — better callbacks with predictable inputs. The signature depends on the event group:
### Task events (`TASK_*`)
```python
from quiv import Event
from quiv.models import Task
def on_task_event(event: Event, task: Task) -> None:
print(f"[{event.value}] Task '{task.task_name}' (id={task.id})")
```
### Job events (`JOB_*`)
```python
from quiv import Event
from quiv.models import Task, Job
def on_job_event(event: Event, task: Task, job: Job) -> None:
print(f"[{event.value}] Job {job.id} for '{task.task_name}'")
if job.duration_seconds is not None:
print(f" Duration: {job.duration_seconds:.2f}s")
if job.error_message is not None:
print(f" Error: {job.error_message}")
```
Async callbacks use the same signatures:
```python
async def on_task_event(event: Event, task: Task) -> None:
...
async def on_job_event(event: Event, task: Task, job: Job) -> None:
...
```
!!! tip "Full type safety"
Since listeners receive `Task` and `Job` model objects, you get IDE autocomplete and type checking on every field — no more guessing dict keys at runtime.
## Registering listeners
Use `add_listener()` to register a callback for a specific event:
```python
from quiv import Quiv, Event
from quiv.models import Task
scheduler = Quiv()
def on_task_added(event: Event, task: Task) -> None:
print(f"Task '{task.task_name}' added with ID {task.id}")
scheduler.add_listener(Event.TASK_ADDED, on_task_added)
```
### Multiple listeners
You can register multiple listeners for the same event. They are called in registration order:
```python
scheduler.add_listener(Event.JOB_FAILED, log_failure)
scheduler.add_listener(Event.JOB_FAILED, send_alert)
```
### Multiple events
Register the same callback for different events within the same event group:
```python
from quiv.models import Task, Job
def job_audit_log(event: Event, task: Task, job: Job) -> None:
print(f"[{event.value}] task={task.task_name} job={job.id}")
scheduler.add_listener(Event.JOB_COMPLETED, job_audit_log)
scheduler.add_listener(Event.JOB_FAILED, job_audit_log)
```
## Removing listeners
Use `remove_listener()` to unregister a previously added callback:
```python
scheduler.remove_listener(Event.TASK_ADDED, on_task_added)
```
If the callback is not found, the call is silently ignored.
## Async listeners
Async listeners run on the main event loop via `run_coroutine_threadsafe`, just like async progress callbacks. This makes them ideal for FastAPI apps where you want to broadcast events to WebSocket clients:
```python
from quiv.models import Task, Job
async def on_job_completed(event: Event, task: Task, job: Job) -> None:
await ws_manager.broadcast({
"type": "job_completed",
"task": task.task_name,
"duration_seconds": job.duration_seconds,
})
scheduler.add_listener(Event.JOB_COMPLETED, on_job_completed)
```
## Error handling
If a listener raises an exception, quiv logs the error but does **not** fail the scheduler or the job. Other listeners for the same event still run. This prevents a broken listener from disrupting task execution.
```mermaid
flowchart TD
A["Event emitted"] --> B["Dispatch listener 1"]
B --> C{Raises?}
C -- No --> D["Dispatch listener 2"]
C -- Yes --> E["Log error"]
E --> D
D --> F["Continue"]
```
## Without an event loop
In scripts without asyncio, sync event listeners work normally — they run directly on the calling thread:
```python
from quiv import Quiv, Event
from quiv.models import Task
scheduler = Quiv()
def on_added(event: Event, task: Task) -> None:
print(f"Added: {task.task_name}")
scheduler.add_listener(Event.TASK_ADDED, on_added)
scheduler.add_task("my-task", lambda: None, interval=10)
# Prints: Added: my-task
```
Async listeners also work in this scenario — they run in a temporary event loop on the calling thread, so `await` calls inside the listener execute correctly.
## FastAPI example
A complete example wiring event listeners into a FastAPI app with WebSocket notifications:
```python
import logging
from contextlib import asynccontextmanager
from typing import Any
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from quiv import Event, Quiv
from quiv.models import Task, Job
scheduler = Quiv(timezone="UTC")
logger = logging.getLogger(__name__)
connected_clients: list[WebSocket] = []
async def broadcast(message: dict) -> None:
for ws in connected_clients:
try:
await ws.send_json(message)
except Exception:
pass
async def on_job_event(event: Event, task: Task, job: Job) -> None:
"""Broadcast job lifecycle events to WebSocket clients."""
payload: dict[str, Any] = {
"event": event.value,
"task_name": task.task_name,
"job_id": job.id,
}
if job.duration_seconds is not None:
payload["duration_seconds"] = job.duration_seconds
if job.error_message is not None:
payload["error"] = job.error_message
await broadcast(payload)
def sync_task() -> None:
pass # your task logic
@asynccontextmanager
async def lifespan(app: FastAPI):
# Register event listeners
scheduler.add_listener(Event.JOB_STARTED, on_job_event)
scheduler.add_listener(Event.JOB_COMPLETED, on_job_event)
scheduler.add_listener(Event.JOB_FAILED, on_job_event)
scheduler.add_listener(Event.JOB_CANCELLED, on_job_event)
scheduler.add_task("my-task", sync_task, interval=60)
scheduler.start()
yield
scheduler.shutdown()
app = FastAPI(lifespan=lifespan)
@app.websocket("/ws/events")
async def events_websocket(websocket: WebSocket):
await websocket.accept()
connected_clients.append(websocket)
try:
while True:
await websocket.receive_text()
except WebSocketDisconnect:
connected_clients.remove(websocket)
```
---
# Cancellation
quiv uses cooperative cancellation via `threading.Event`. This gives handlers full control over how and when they stop, enabling clean resource cleanup and graceful shutdown.
## How it works
Each job gets its own `threading.Event` stop signal. When cancellation is requested, the event is set. The handler checks the event and decides how to respond.
```mermaid
sequenceDiagram
participant App as Application
participant S as Scheduler
participant H as Handler Thread
S->>H: Submit job (with stop_event)
H->>H: Working...
App->>S: cancel_job(job_id)
S->>S: stop_event.set()
H->>H: Checks _stop_event.is_set()
H->>H: Cleans up and returns
S->>S: Detects stop_event was set
S->>S: Mark job as "cancelled"
```
### Key points
- Cancellation is **cooperative** — the handler must check `_stop_event` to actually stop. quiv cannot force-kill a running thread.
- The handler decides **when** to check and **how** to clean up.
- Job status is automatically set to `cancelled` if the stop event was set when the handler returns, regardless of whether the handler accepted `_stop_event` in its signature.
## Writing a cancellable handler
Add `_stop_event` to your handler's signature. quiv inspects the signature and only injects it if the parameter is present.
```python
import threading
import time
def long_running_task(
_stop_event: threading.Event | None = None,
):
"""A task that processes items and can be cancelled between steps."""
items = fetch_items()
for item in items:
if _stop_event and _stop_event.is_set():
# Clean up and exit gracefully
return
process(item)
time.sleep(0.1)
```
### Check frequency
Check `_stop_event` at natural breakpoints in your handler:
- Between iterations of a loop
- Before starting an expensive operation
- After completing a unit of work
```python
def batch_processor(
_stop_event: threading.Event | None = None,
_progress_hook: Callable | None = None,
):
batches = get_batches()
for i, batch in enumerate(batches):
# Check before each batch
if _stop_event and _stop_event.is_set():
return
result = process_batch(batch) # might take a while
save_result(result)
if _progress_hook:
_progress_hook(step=i + 1, total=len(batches))
```
### Using `_stop_event.wait()` instead of `time.sleep()`
If your handler has a sleep/wait period, use `_stop_event.wait()` instead of `time.sleep()`. This makes cancellation responsive even during wait periods:
```python
def polling_task(
_stop_event: threading.Event | None = None,
):
"""Poll an API every 5 seconds, but respond to cancellation immediately."""
while not (_stop_event and _stop_event.is_set()):
result = check_api()
if result.ready:
handle_result(result)
return
# Wait 5 seconds OR until cancelled — whichever comes first
if _stop_event:
_stop_event.wait(timeout=5)
else:
time.sleep(5)
```
This pattern is especially useful for tasks that poll external services.
## Cancelling a job
Use `cancel_job(job_id)` to signal cancellation:
```python
# Find running jobs
jobs = scheduler.get_all_jobs(status="running")
# Cancel a specific job
for job in jobs:
scheduler.cancel_job(job.id)
```
`cancel_job` returns `True` if the stop event was found and set, `False` if the job was not found (already finished or invalid ID).
## Cancellation during shutdown
When `shutdown()` is called, quiv automatically cancels all tracked running jobs by setting their stop events. Handlers that check `_stop_event` will exit gracefully; handlers that don't will run to completion before the process exits.
```mermaid
flowchart TD
A["shutdown() called"] --> B[Set _shutdown flag]
B --> C[Cancel all tracked jobs]
C --> D[Join scheduler thread]
D --> E[Shutdown thread pool]
E --> F[Dispose DB engine]
F --> G["Delete DB files (.db, -wal, -shm)"]
```
## How status is determined
When a job finishes, quiv checks the stop event to determine the final status. This happens in the `finally` block of `_run_job`, so it works regardless of how the handler exited:
```mermaid
flowchart TD
A[Handler returns] --> B{Exception raised?}
B -- Yes --> C["status = failed"]
B -- No --> D["status = completed"]
C --> E{"stop_event.is_set()?"}
D --> E
E -- Yes --> F["status = cancelled (overrides failed/completed)"]
E -- No --> G[Keep current status]
F --> H[Finalize job in DB]
G --> H
```
Note that `cancelled` takes priority over both `completed` and `failed`. If a handler raises an exception *and* the stop event is set, the job is marked as `cancelled` — the assumption is that the cancellation caused the error.
## Handlers without `_stop_event`
If your handler's signature does not include `_stop_event` (and does not use `**kwargs`), the event is **not injected** — but it is still tracked internally. This means:
- `cancel_job()` still sets the event
- The job status is still set to `cancelled` if the event was set when the handler returns
- The handler itself just can't respond to cancellation early
This is useful for short-lived tasks where you don't need mid-execution cancellation but still want correct status tracking on shutdown.
## Combining with progress callbacks
A common pattern is to check `_stop_event` and report progress in the same loop:
```python
def export_data(
format: str,
_stop_event: threading.Event | None = None,
_progress_hook: Callable | None = None,
):
records = query_records()
total = len(records)
for i, record in enumerate(records, 1):
if _stop_event and _stop_event.is_set():
return
write_record(record, format)
if _progress_hook and i % 100 == 0:
_progress_hook(
step=i,
total=total,
pct=round(i / total * 100),
)
```
The progress callback and stop event are independent — you can use either or both. quiv injects each one only if the handler's signature accepts it.
---
# Exceptions
All custom exceptions inherit from `QuivError`.
## Hierarchy
- `QuivError`
- `ConfigurationError`
- `InvalidTimezoneError`
- `DatabaseInitializationError`
- `HandlerRegistrationError`
- `HandlerNotRegisteredError`
- `TaskNotScheduledError`
- `TaskNotActiveError`
- `TaskNotFoundError`
- `JobNotFoundError`
## Exception reference
### `ConfigurationError`
Raised when runtime or scheduling configuration is invalid, for example:
- `pool_size <= 0`
- `history_retention_seconds < 0`
- invalid `add_task(...)` inputs (`task_name`, `interval`, `delay`)
- mixing `config=...` with direct constructor config args
### `InvalidTimezoneError`
Raised when timezone input is not a valid IANA timezone or not a `str/tzinfo`.
### `DatabaseInitializationError`
Raised when SQLite/SQLModel initialization fails during scheduler creation.
### `HandlerRegistrationError`
Raised when registering invalid handlers/callbacks (empty task id, non-callable handler/callback).
### `HandlerNotRegisteredError`
Raised when an operation requires a registered handler but none exists for the given task id[^1].
[^1]: This typically means `run_task_immediately()` was called with a `task_id` that was never returned by `add_task()`, or the task was already removed.
### `TaskNotScheduledError`
Raised when a task handler is registered but the scheduled task row no longer exists in the database, for example if it was deleted externally.
### `TaskNotActiveError`
Raised when an operation requires an `active` task. Currently raised by `run_task_immediately()` when the task is `running` (a second concurrent run would break the no-overlap guarantee) or `paused` (un-pausing must be an explicit `resume_task()` call).
### `TaskNotFoundError`
Raised when a task ID lookup fails in persistence operations (pause/resume/finalize).
### `JobNotFoundError`
Raised when a job ID lookup fails in persistence operations (mark running/finalize).
## Handling pattern
```python
from quiv import Quiv
from quiv.exceptions import QuivError, ConfigurationError
try:
scheduler = Quiv(pool_size=4, timezone="UTC")
except ConfigurationError as exc:
print("bad config", exc)
except QuivError as exc:
print("scheduler init failed", exc)
```
For application boundaries, catch `QuivError` to cover all library-specific failures, and optionally catch specific subclasses when you need targeted recovery.
---
# Testing
quiv is extensively tested with **108 tests** covering the full lifecycle of tasks, jobs, event listeners, progress callbacks, configuration, models, and edge cases. Tests run on every commit via CI across Python 3.10 through 3.14.
```bash
# Run all tests
uv run pytest
# Run with coverage
uv run pytest --cov=quiv
# Run a specific test file
uv run pytest tests/test_scheduler.py
# Run a single test
uv run pytest tests/test_scheduler.py::test_backpressure_skips_dispatch_when_pool_full
```
## Test architecture
Most tests require a running asyncio event loop for callback dispatch. The `running_main_loop` fixture (in `conftest.py`) spins up an event loop in a background thread and yields it to each test. Every test calls `scheduler.shutdown()` in a `finally` block to clean up threads and temp DB files.
## What is tested
### Scheduler lifecycle and configuration
- Mixing `config=QuivConfig(...)` with explicit kwargs raises `ConfigurationError`
- `pool_size <= 0` and `history_retention_seconds < 0` are rejected
- `start()` is idempotent (safe to call multiple times)
- `shutdown()` handles DB cleanup failures gracefully
- Database initialization failure raises `DatabaseInitializationError`
- quiv's internal tables (`quiv_task`, `quiv_job`) do not leak into user SQLModel metadata
### Task input validation
- Empty `task_name` is rejected
- `interval <= 0` and `delay < 0` are rejected
- `args` must be a `tuple` (not list)
- `kwargs` must be a `dict` (not string)
- Unpicklable args (e.g. lambdas) raise `ConfigurationError` with a clear message
### Task registration and identification
- `add_task()` returns a unique `task_id` (UUID)
- Duplicate `task_name` values are allowed, each getting a distinct `task_id`
- Handlers, progress callbacks, and DB rows are all keyed by `task_id`
- `remove_task()` cleans up handler, progress callback, and DB row
- Removing a non-existent task raises `TaskNotFoundError`
- Deleting a non-existent task from persistence raises `TaskNotFoundError`
### Task execution
- Sync run-once task executes and produces a `completed` job
- Async run-once task executes via thread-local event loop
- Sync handler without `_stop_event` or `_progress_hook` still runs correctly
- Failed handler sets job status to `failed`
- `_job_id` is injected as a UUID string when handler accepts it
- `args` and `kwargs` ordering is preserved through pickle round-trip (tested with 8 positional args and 5 keyword args)
### Concurrent execution and backpressure
- Same task is never dispatched concurrently (status set to `running` blocks re-dispatch)
- When the thread pool is full, due tasks are deferred to the next tick instead of queued unboundedly
- Deferred tasks execute once a worker becomes available
- `_active_job_count` decrements correctly after job completion
- Late-starting jobs (due to pool saturation) log a warning with the delay
### Interval scheduling (`fixed_interval`)
- **Fixed interval** (`fixed_interval=True`): next run is aligned to `start_time + interval`
- **Skipped intervals**: a 70-second job with 60-second interval skips to `start_time + 120s`; a 130-second job skips to `start_time + 180s`
- **Wait between runs** (`fixed_interval=False`): next run is `completion_time + interval`
- Recurring task finalization sets status back to `active` and updates `next_run_at`
- Run-once task finalization deletes the task row
### Cancellation
- `cancel_job()` returns `True` when stop event exists, `False` otherwise
- Handler that sets `_stop_event` results in `cancelled` status
- `remove_task()` on a running task cancels its active job
- `shutdown()` cancels all tracked running jobs
### Progress callbacks
- Async progress callback dispatched on main event loop via handler's `_progress_hook`
- Sync progress callback dispatched on main event loop via `call_soon_threadsafe`
- Async handler with sync progress callback works correctly
- Progress callback registration and clearing via `None`
- Sync callback works without an event loop (runs on worker thread)
- Async callback works without an event loop (runs in temporary event loop)
- Failing sync callback is logged, does not crash the job
- Failing async callback is logged, does not crash the job
- Closed main loop does not crash progress dispatch
### Event listeners
- Invalid event type (non-`Event` enum) raises `ConfigurationError`
- Non-callable callback raises `ConfigurationError`
- Removing an unregistered listener is silently ignored
- **`TASK_ADDED`**: listener receives `Event` and `Task` with correct `task_name` and `task_id`
- **`TASK_REMOVED`**: listener receives snapshot of task before deletion
- **`TASK_PAUSED`**: listener receives task with `paused` status
- **`TASK_RESUMED`**: listener receives task with `active` status
- **`JOB_STARTED`**: listener receives `Task` and `Job` with `running` status
- **`JOB_COMPLETED`**: listener receives `Job` with `duration_seconds` set and `error_message` as `None`
- **`JOB_FAILED`**: listener receives `Job` with `error_message` matching the exception
- **`JOB_CANCELLED`**: listener receives `Job` with `cancelled` status
- Multiple listeners for the same event are all called
- Async listener dispatched on main event loop
- Failing listener is logged and swallowed; subsequent listeners still run
- Sync listener works without an event loop
- Async listener works without an event loop (temporary event loop)
- Async listener failure without an event loop is caught and logged
### Handler injection
- `_job_id`, `_stop_event`, and `_progress_hook` are injected when handler accepts them
- Injection is skipped when handler signature does not include the parameters
- Handlers with `**kwargs` receive all injected parameters
- Uninspectable callables (e.g. `object()`) are handled gracefully (no injection, no crash)
### Deserialization safety
- Corrupt pickle data in `args_pickled` raises `ConfigurationError`
- Corrupt pickle data in `kwargs_pickled` raises `ConfigurationError`
- Pickled kwargs that aren't a `dict` raises `ConfigurationError`
- Corrupt pickle in `Task.model_validate()` from dict falls back to empty defaults
- Corrupt pickle in `Task.model_validate()` from `TaskDB` object falls back to empty defaults
- Non-standard input types pass through the validator without crashing
### Datetime normalization
- Naive datetimes are normalized to UTC-aware (treated as UTC)
- Timezone-aware datetimes are converted to UTC
- `None` datetimes pass through unchanged
- `Task` public model normalizes `next_run_at` from `TaskDB`
- `TaskDB` datetimes are normalized on DB load via `@reconstructor`
- `Job` datetimes (`started_at`, `ended_at`) are normalized on DB load via `@reconstructor`
- `Job` with `None` `ended_at` is handled correctly
- `get_all_tasks()` returns UTC-aware `next_run_at` regardless of configured display timezone
- `get_job()` returns UTC-aware `started_at` and `ended_at`
### Persistence layer
- `queue_task_for_immediate_run()` raises `TaskNotScheduledError` for missing task
- `pause_task()` and `resume_task()` raise `TaskNotFoundError` for missing task
- `mark_task_running()` raises `TaskNotFoundError` for missing task
- `mark_job_running()` and `finalize_job()` raise `JobNotFoundError` for missing job
- History cleanup deletes old finished jobs while keeping recent ones
- Job status filtering (`completed`, `failed`) returns correct subsets
- `get_all_tasks(include_run_once=False)` excludes run-once tasks
- Paused tasks are excluded from due-task queries
- Resumed tasks appear in due-task queries
### Immediate execution
- `run_task_immediately()` raises `HandlerNotRegisteredError` for unregistered handler
- `run_task_immediately()` raises `TaskNotScheduledError` when task row is missing
- `run_task_immediately()` successfully queues a registered task
### Configuration
- IANA timezone string resolves correctly
- `tzinfo` instance passes through
- Invalid timezone string raises `InvalidTimezoneError`
- Invalid type raises `InvalidTimezoneError`
- `QuivConfig` works without conflict when no explicit kwargs are passed
### Scheduler loop resilience
- Loop catches and logs exceptions, retries after sleep
- Loop does not crash on persistent errors (verified over multiple iterations)
### Model serialization
- `TaskDB` field serializer unpickles valid bytes for JSON output
- `Task` model serializes args/kwargs as unpickled Python objects
- JSON serialization produces correct output (tested with `model_dump_json()`)
- Time helpers (`next_run_time`, `get_current_time`) return UTC-aware datetimes
- `id_generator()` returns UUID strings