API
Worker and handlers
Task registry, JobContext, and running workers.
Task + registry
from pydantic import BaseModel
from taskq import Task, TaskRegistry
class Input(BaseModel):
n: int
class Output(BaseModel):
n: int
async def work(payload: Input) -> Output:
return Output(n=payload.n)
TASK = Task(
name="demo.work",
queue="demo",
input_model=Input,
output_model=Output,
handler=work,
retry=5, # bool | int | RetryStrategy
)
registry = TaskRegistry((TASK,))
Wire names match [a-z][a-z0-9_]*(?:\.[a-z][a-z0-9_]*)*. Queues match [a-z0-9_]{1,57}.
JobContext
Two-argument handlers receive JobContext first:
from taskq import Complete, JobContext
async def work(ctx: JobContext, payload: Input) -> Complete:
ctx.raise_if_cancelled()
await ctx.checkpoint({"last_item": payload.n})
return Complete(result={})
JobContext exposes cancellation checks, the latest progress checkpoint,
checkpoint(...), run_sync(...), and (when configured) report_effect(...).
It deliberately does not expose job_id or attempt_id; fencing identity
stays inside the worker runtime.
Running workers
Prefer the CLI (taskq --context NAME worker run --registry module:attr …). Programmatically, build a WorkerService / WorkerSupervisor with a transport + registry — see the package worker module and Stage-2 specs in the repo.
| Concern | API |
|---|---|
| Env / flags | WorkerSettings (TASKQ_*) |
| Process loop | WorkerService |
| Fair poll + presence | WorkerSupervisor |