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.job_id, ctx.attempt_id, cancellation token, …
return Complete(result={})
Running workers
Prefer the CLI (taskq worker --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 |