Getting Started
Quickstart
Migrate, register a task, enqueue, and run a worker.
1. Migrate
Create ~/.config/taskq/config.toml and export the credential:
version = 1
[contexts.development]
transport = "sql"
dsn_env = "TASKQ_DEVELOPMENT_DSN"
expected_environment = "development"
actor = "operator:developer"
export TASKQ_DEVELOPMENT_DSN='postgresql://postgres:postgres@localhost:5432/myapp'
taskq --context development db plan -o json
taskq --context development --yes db migrate --plan-digest "$PLAN_DIGEST"
taskq --context development target show -o json
taskq --context development --yes \
--expected-installation-id "$TASKQ_INSTALLATION_ID" \
target bind development --expected-binding-version 0
taskq --context development db plan -o json
taskq --context development --yes db migrate --plan-digest "$PLAN_DIGEST"
taskq --context development db verify
Extract each fresh PLAN_DIGEST and the installation UUID from the preceding
JSON envelope after reviewing it. Migration stops safely at the unbound-target
gate; the second plan resumes from that point.
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 --context development worker run \
--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.