Getting Started
Quickstart
Migrate, register a task, enqueue, and run a worker.
1. Migrate
taskq migrate "postgresql://postgres:postgres@localhost:5432/myapp"
taskq verify "postgresql://postgres:postgres@localhost:5432/myapp"
2. Define a task
tasks.py
from pydantic import BaseModel
from taskq import Complete, Task, TaskRegistry
class Input(BaseModel):
value: int
class Output(BaseModel):
doubled: int
async def double(payload: Input) -> Output:
return Output(doubled=payload.value * 2)
DOUBLE = Task(
name="demo.double",
queue="demo",
input_model=Input,
output_model=Output,
handler=double,
)
registry = TaskRegistry((DOUBLE,))
Handlers take (payload) or (JobContext, payload). Returning the output model is treated as success; you can also return Complete, Snooze, Cancel, Retry, or NonRetryable explicitly.
3. Enqueue
from taskq import TaskQ
tq = TaskQ.from_dsn(
"postgresql+asyncpg://postgres:postgres@localhost:5432/myapp",
registry=registry,
)
result = await tq.enqueue(DOUBLE, {"value": 3}, idempotency_key="demo:3")
# result.status in {"created", "existed"}
# result.created is True on first insert
await tq.aclose()
TaskQ is a producer-facing facade. Configuring a registry does not start a claim loop.
4. Run a worker
taskq worker \
--dsn "postgresql://postgres:postgres@localhost:5432/myapp" \
--registry tasks:registry \
--queue demo \
--environment development
Point --registry at a module:attr that exports a TaskRegistry. For local smoke tests without NOTIFY, add --no-listen --poll-interval 0.5.