Work after the response
Background Jobs
Send notifications, call integrations, or rebuild a search index outside the request — through named
handlers on a bounded queue with retries, timeouts, and graceful shutdown. Not
asyncio.create_task, threads, or FastAPI BackgroundTasks.
Choose a driver
| JOBS_DRIVER | Survives a crash or restart | Shared by |
|---|---|---|
| memory (default) | No — queued and running jobs are lost | One process |
| sqlite | Yes — stored in storage/jobs.sqlite3 |
Every process on one machine |
Neither driver coordinates separate machines. When several hosts must share one queue, use an external broker.
Defining jobs
Put jobs in src/lib/jobs.py. main.py imports it at startup and only starts workers when at least one job is registered.
# src/lib/jobs.py
from casp.jobs import JobError, current_job, job
@job(attempts=3, backoff=(0.25, 30), timeout=20)
async def send_receipt(order_id: int):
order = await prisma.order.find_unique(where={"id": order_id})
if order is None:
raise JobError.discard(f"order {order_id} no longer exists")
try:
await mailer.send_receipt(order)
except ProviderUnavailable as exc:
raise JobError.retry(str(exc))
@job("search.reindex", timeout=None)
def reindex(model: str):
# Sync handlers run in a worker thread.
search_index.rebuild(model)
| Option | Default | Meaning |
|---|---|---|
| name | function name, _ → - |
1–64 of letters, digits, . _ -; unique |
| attempts | 3 | Total attempts including the first |
| backoff | (0.25, 30) | Initial and maximum retry delay, doubling each attempt |
| timeout | 30 | Seconds per attempt; None for no limit |
Dispatching
Dispatch after the authoritative write commits, so a fast worker never sees data that later rolls back.
order = await prisma.order.create(data=...)
await send_receipt.dispatch(order_id=order.id)
# or by name
from casp import jobs
job_id = await jobs.dispatch("search.reindex", model="Product")
-
Payloads are JSON-serializable keyword arguments; a bad payload raises
TypeErrorat dispatch. - The handler receives only the parameters it declares.
-
try_dispatchnever waits and raisesQueueFull; during shutdown dispatch raisesQueueClosed. - Pass an id, not sessions, secrets, or personal data. Jobs do not inherit the request's session — authorize before dispatching.
Retries, timeouts, and idempotency
returns
Succeeded.
JobError.discard
Discarded immediately.
retry / any exception
Retried with backoff.
exceeds timeout
Counted as timed out, then retried.
attempts used
Failed (no dead-letter store in memory).
A timeout cancels the handler after its side effect may already have happened. Every retryable
handler must be idempotent — use a database uniqueness key, a provider idempotency key, or a completion marker.
Inside a handler, current_job() returns the id, name, attempt, max attempts, and the dispatching request id.
The SQLite driver
JOBS_DRIVER=sqlite JOBS_DATABASE=storage/jobs.sqlite3 # default JOBS_POLL_INTERVAL=1 # seconds between checks for new work
-
A job is written before
dispatchreturns;QueueFullis raised pastJOBS_CAPACITY(10,000). - Workers claim one job at a time atomically, so processes sharing the file never run a job twice.
- A running job holds a renewed five-minute lease; if the process dies another worker picks it up (that attempt counts).
-
Succeeded jobs are deleted; failed and discarded ones stay for inspection via
get_queue().jobs("failed")andpurge(older_than=...). -
storage/is not deleted bynpm run dev. It is gitignored — keep it on a persistent volume in container deployments, and keepjobs()off public endpoints because it returns stored payloads.
Configuration and shutdown
| Variable | Default | Meaning |
|---|---|---|
| JOBS_DRIVER | memory | memory or sqlite |
| JOBS_WORKERS | 1 | Concurrent worker tasks per process |
| JOBS_CAPACITY | 1024 / 10000 | Queue bound (memory / sqlite) |
| JOBS_DATABASE | storage/jobs.sqlite3 | SQLite file |
| JOBS_POLL_INTERVAL | 1 | SQLite polling interval in seconds |
| JOBS_SHUTDOWN_GRACE | 30 | Seconds to drain on shutdown |
Workers start and stop with the app. The job lifespan starts after the database lifespan and exits before it, so handlers can use
Prisma until they finish. Each outcome is logged with target jobs — payload values never are — and
casp.jobs.metrics() returns process-local counters. To test, call the job directly
(await send_receipt(order_id=1)) or install a fresh queue with configure(JobQueue(...)).
Task Scheduling
Dispatch these jobs every few minutes, daily at a local time, or on a cron expression.