Skip to content

CLI reference

The taskq CLI is the primary operational interface for managing migrations, running workers, probing health, and serving the admin UI. All commands load settings from TASKQ_* environment variables or .env files via dotenvmodel; environment variables take precedence over .env files.

Installation

The taskq command is installed as part of the taskq-py package:

uv add taskq-py

The console-script entry point is taskq.cli:main. See ../getting-started/quick-start.md for initial environment setup.


Global environment variables

These variables are read by all commands via TaskQSettings.load() or WorkerSettings.load().

Variable Default Description
TASKQ_PG_DSN postgresql://taskq:taskq@localhost:5432/taskq Postgres connection string used by migrate and health commands
TASKQ_SCHEMA_NAME taskq Postgres schema for all TaskQ tables

Worker-specific variables (pool sizes, heartbeat timing, cancellation grace periods, etc.) are documented in workers.md.


taskq dev

Starts a worker in development mode with automatic restart on file changes. Useful during local development — saves manually killing and restarting the worker after every code edit.

taskq dev MODULE:ATTR [OPTIONS]

Arguments:

Argument Description
MODULE:ATTR module:attr reference to the actor registry — same syntax as taskq worker --actors.

Options:

Option Type Default Description
--watch str (repeatable) current working directory Path to watch for changes. Pass the flag multiple times to watch several directories.
--grace-period int 5 Seconds between SIGTERM and SIGKILL when stopping the worker before a restart. 0 means send SIGTERM then kill immediately if still alive.

How it works:

  1. Imports MODULE:ATTR at startup and exits with a clear error if the module is not found or the attribute does not exist.
  2. Spawns taskq worker --actors MODULE:ATTR as a child subprocess. The child inherits the parent's full environment, so TASKQ_* variables set in the shell are picked up normally.
  3. Watches the specified paths with watchfiles.awatch() using DefaultFilter (ignores .git, __pycache__, editor temp files, etc.) and a 400 ms debounce window.
  4. On any file change: sends SIGTERM to the running worker, waits up to --grace-period seconds, then SIGKILL if still alive. Re-validates the import, then spawns a fresh worker. If the re-import fails (e.g. a syntax error) it logs a warning and waits for the next change rather than crashing.
  5. Ctrl-C (SIGINT) stops the worker cleanly and exits with code 0.

Each restart gets a fresh Python interpreter — no importlib.reload() is involved, so module-level state (Pydantic models, actor registries, config objects) is always clean.

Requirements:

taskq dev requires the reload extra:

uv add "taskq-py[reload]"

This installs watchfiles. Without it the command prints an install hint and exits with code 1.

Example: watch a single package directory

taskq dev myapp.actors:registry --watch src/myapp

Example: watch multiple directories

taskq dev myapp.actors:registry --watch src/myapp --watch config

Example: fast restart (no grace period)

taskq dev myapp.actors:registry --grace-period 0

Exit codes:

Code Condition
0 Clean exit via Ctrl-C
1 Bad MODULE:ATTR syntax, import failure, or watchfiles not installed

Note: taskq dev is intended for local development only. Do not run it in production — use taskq worker directly under a process supervisor (systemd, Docker, Kubernetes).


taskq migrate status

Shows which migrations have been applied and which are pending.

taskq migrate status

Connects to TASKQ_PG_DSN, queries {schema}.schema_migrations, and prints one line per discovered migration file.

Example output:

schema: taskq
applied: 1
  [✔] 01.00.00_01_pre_initial.sql

A marker indicates the migration has been applied. An empty marker indicates it is pending. Migrations are discovered from the built-in migration directory in the package.

No options. Uses TASKQ_PG_DSN and TASKQ_SCHEMA_NAME from the environment.


taskq migrate up

Applies pending migrations in order.

taskq migrate up [OPTIONS]

Options:

Option Type Default Description
--phase pre \| post \| None None Restrict to only pre or only post phase migrations. When absent, applies both phases in order.
--target str \| None None Stop after applying this migration version (inclusive). Version format matches the filename prefix, e.g. 01.00.00_01.
--max-steps int \| None None Maximum number of migrations to apply in this invocation.

The command is idempotent: each migration is recorded in {schema}.schema_migrations and is skipped on subsequent runs. Running taskq migrate up with no options applies all pending migrations.

On failure (exit code 1) the command never prints a traceback. When a migration fails during apply, the command diagnoses itself: it names the failed migration, reports the state the schema was left in — a clean rollback for transactional migrations, or for -- taskq:no-transaction migrations the statements that remain applied plus any INVALID indexes found — and prints the single action to take. When the database connection itself cannot be established, there is nothing to diagnose, so it prints a short report naming the connection error and the same re-run action.

Example: apply all pending:

taskq migrate up

Example output:

applied 2 migration(s):
  01.01.00_01_pre_add_reservation_slots.sql
  01.01.00_02_post_add_reservation_slots.sql

When there is nothing to apply:

no pending migrations

Example: apply only pre-phase migrations up to a target version:

taskq migrate up --phase pre --target 01.01.00_01

taskq worker

Starts a TaskQ worker process. Blocks until SIGTERM or SIGINT.

taskq worker --actors MODULE:ATTR [OPTIONS]

Options:

| Option | Type | Default | Env var override | Description | |---|---|---|---|---|---| | --actors | str | required | — | module:attr reference to the actor registry | | --queues | list[str] | None | TASKQ_QUEUES | Queue names to consume; repeat the flag once per queue name | | --max-concurrency | int | None | TASKQ_MAX_CONCURRENCY | Upper bound on concurrent jobs | | --poll-interval | float | None | TASKQ_POLL_INTERVAL | Producer loop fallback polling cadence (seconds) | | --worker-group | str | None | TASKQ_WORKER_GROUP | Consumer group name for observability spans | | --worker-label | str | None | TASKQ_WORKER_LABEL | Human-readable label stored in the workers table | | --workgroup-instance | str | None | TASKQ_WORKGROUP_INSTANCE | UUIDv7 identifying the workgroup orchestrator that launched this worker | | --health-socket-path | str | None | TASKQ_HEALTH_SOCKET_PATH | Unix socket path for the health server (use unique paths when running multiple workers) | | --force-update-actor-config | bool | False | TASKQ_FORCE_UPDATE_ACTOR_CONFIG | Overwrite drifted actor-config rows at startup | | --until-idle | bool | False | — | Run until all subscribed queues are drained, then exit. Exit 0 if all jobs succeeded, 3 if any failed, 4 if idle-max-runtime was exceeded. Incompatible with cron-driven workloads. | | --idle-settle-window | float | None | TASKQ_IDLE_SETTLE_WINDOW | Seconds to wait after queues appear empty before declaring drained. Default 2.0. Only used with --until-idle. | | --idle-poll-interval | float | None | TASKQ_IDLE_POLL_INTERVAL | How often to check queue depth. Default 1.0. Only used with --until-idle. | | --idle-max-runtime | float | None | TASKQ_IDLE_MAX_RUNTIME | Maximum wall-clock seconds before forcing exit (code 4). Only used with --until-idle. |

All other worker settings are read from environment variables. See workers.md for the full list.

--actors format

The value must be a module:attr string with exactly one colon separator. Both module and attr must be non-empty.

myapp.actors:registry          # module=myapp.actors, attr=registry
myapp.workers.email:handlers   # module=myapp.workers.email, attr=handlers

The module is imported at startup via importlib.import_module. The attribute must resolve to one of:

  • Mapping[str, ActorRef] — keys are actor names, values are ActorRef instances.
  • Iterable[ActorRef] — names are read from ActorRef.name on each element.

Any other type (including a plain list of non-ActorRef objects) prints an error and exits with code 1.

Example: iterable form

# myapp/actors.py
from taskq.actor import actor
from pydantic import BaseModel


class EmailPayload(BaseModel):
    to: str
    subject: str


@actor(queue="email")
async def send_email(payload: EmailPayload) -> None: ...


registry = [send_email]
taskq worker --actors myapp.actors:registry

Example: mapping form

# myapp/actors.py
registry = {"send_email": send_email, "resize_image": resize_image}
taskq worker --actors myapp.actors:registry

--queues flag

--queues is a multi-value flag. Pass it once for each queue name:

taskq worker --actors myapp.actors:registry --queues default --queues priority --queues email

Do not pass a comma-separated string directly to --queues on the command line. Use the environment variable for comma-separated input:

TASKQ_QUEUES=default,priority,email taskq worker --actors myapp.actors:registry

On the command line the flag must be repeated once per queue.

--force-update-actor-config

At startup the worker compares each registered actor's queue and metadata values against the stored row in {schema}.actor_config. If either differs and this flag is absent, the worker refuses to start and prints:

ActorConfigDriftList: ...
Re-run with --force-update-actor-config to overwrite, or set TASKQ_FORCE_UPDATE_ACTOR_CONFIG=true.

max_concurrent, max_pending, and result_ttl never appear in this error — they are operator-owned once a stored row exists (the @actor(...) literal only seeds them the first time) and can never drift. Use taskq actor-config set to change those; see ActorConfig sync.

Use this flag on the first new pod of a rolling deploy when an actor's queue or metadata has changed. Remove it for subsequent pods — it is not safe to run permanently as it allows silent structural drift. See workers.md for the full drift protocol.

taskq actor-config

Inspect and tune stored {schema}.actor_config capacity fields without a code change or a worker restart:

taskq actor-config list
taskq actor-config get checkout_charge
taskq actor-config set checkout_charge --max-concurrent 20
taskq actor-config set checkout_charge --max-pending 5000
taskq actor-config set checkout_charge --clear-max-pending

All three capacity fields are live. --max-concurrent and --result-ttl take effect immediately for every worker in the fleet — the dispatch query re-reads max_concurrent every dispatch cycle, and the terminal-write path re-reads result_ttl on every job completion. --max-pending is enforced by every enqueue-side process (your web app, your workers' sub-job enqueues) through a TTL-bounded cache of the table (default 5s staleness), so a change propagates fleet-wide within seconds — no redeploy, no worker restart.

Clearing a field writes NULL, which means different things per field on purpose: --clear-max-concurrent makes the actor unlimited (the dispatch SQL cannot see the code literal once a row exists), while --clear-max-pending / --clear-result-ttl revert to the @actor(...) literal — clearing undoes your override and restores the code default. For result_ttl the reverted literal is applied from each job's completion timestamp by the completing worker (never re-pinned to the enqueue time), so a job that sat in the queue longer than its TTL still gets its full result lifetime.

To see why a change is (or isn't) taking effect, diff the stored rows against the code literals in your registry:

taskq actor-config diff --actors myapp.actors:registry

Per actor and capacity field this prints the @actor(...) literal, the stored value, and the value the engine currently enforces (effective). It also flags queue/metadata mismatches — those are structural and block the next worker startup with ActorConfigDriftList — actors with no stored row yet, and leftover rows whose actor is no longer registered. Note the no-row case is field-dependent: an actor with no stored row does not dispatch at all (max_concurrent shows effective=0 — the dispatch capacity gate reads only actor_config rows until a worker startup seeds it), while max_pending / result_ttl fall back to the code literal.

taskq actor-config set requires the actor to have a stored row already (created by a worker startup that registered it). queue and metadata are structural and are only ever changed by redeploying with a new @actor(...) registration (plus --force-update-actor-config if a stored row already exists).

taskq actor-config deregister

Deregister an actor: delete its actor_config row with safety checks.

taskq actor-config deregister <ACTOR> [--force] [--purge-queue]
  • <ACTOR> — actor name (positional argument)
  • --force — cancel pending/scheduled jobs, disable enabled cron schedules, and proceed despite non-terminal jobs. Running jobs still block.
  • --purge-queue — also delete the orphaned queues row if no other actor references it.

Exit code 0 on success, 1 on refusal (with error message) or not found.

Exit codes

Code Meaning
0 Clean shutdown (SIGTERM received and shutdown completed)
1 Startup failure or runtime error
3 --until-idle: some jobs failed during the drain
4 --until-idle: idle-max-runtime exceeded before drain completed

ActorConfigDriftList is caught by the CLI and produces a clean one-line error message on stderr. Other bootstrap failures (import errors, wrong attribute type, etc.) produce a Python traceback on stderr. Both exit with code 1.


taskq health live

Probes the worker's liveness endpoint.

taskq health live

Connects to the Unix socket at TASKQ_HEALTH_SOCKET_PATH (default /tmp/taskq_health.sock) and issues GET /live. The health server schedules a loop.call_later(0.01, ...) callback and waits up to 1.0s for it to fire, confirming the event loop is responsive.

The Unix socket is not reachable via Kubernetes httpGet probes. Use an exec probe:

livenessProbe:
  exec:
    command: ["taskq", "health", "live"]
  initialDelaySeconds: 5
  periodSeconds: 10

Exit codes:

Code Condition
0 HTTP 2xx — event loop is responsive ({"status":"ok"})
1 HTTP 5xx — event loop unresponsive or timeout exceeded
1 Socket unreachable (worker not running or wrong socket path)

Example:

taskq health live && echo "worker is alive"

taskq health ready

Probes the worker's readiness endpoint.

taskq health ready

Connects to the Unix socket and issues GET /ready. The health server checks:

  1. shutdown_phase == NONE (worker is not draining or shutting down)
  2. PG ping succeeds within TASKQ_HEALTH_PG_PING_TIMEOUT (default 0.2s)

Both conditions must pass for the response to be 200. During any shutdown phase, the response is 503 regardless of the PG ping result.

Response body (200):

{
  "ready": true,
  "redis_configured": false,
  "active_jobs": 4,
  "is_leader": false,
  "shutdown_phase": null
}

Response body (503):

{
  "ready": false,
  "redis_configured": false,
  "active_jobs": 2,
  "is_leader": false,
  "shutdown_phase": 1
}

shutdown_phase is null when NONE; otherwise the integer value (1=DRAINING, 2=CANCELLING, 3=FORCING, 4=ABANDONING).

Exit codes:

Code Condition
0 HTTP 200 — worker is ready
1 HTTP 503 — not ready (shutting down or PG ping failed)
1 Socket unreachable

Example (Kubernetes readiness probe via exec):

readinessProbe:
  exec:
    command: ["taskq", "health", "ready"]
  initialDelaySeconds: 5
  periodSeconds: 10

taskq health metrics

Fetches the worker's Prometheus-format metrics.

taskq health metrics

Connects to the Unix socket and issues GET /metrics. Always returns 200. Prints the response body to stdout.

Example output:

# HELP taskq_active_jobs Currently in-flight jobs on this worker.
# TYPE taskq_active_jobs gauge
taskq_active_jobs 3
# HELP taskq_is_leader 1 if this worker holds the maintenance leader lock.
# TYPE taskq_is_leader gauge
taskq_is_leader 0
# HELP taskq_shutdown_phase Current shutdown phase enum value (0=NONE).
# TYPE taskq_shutdown_phase gauge
taskq_shutdown_phase 0

Exit codes:

Code Condition
0 HTTP 200 — metrics returned
1 Socket unreachable or request timed out

taskq ui serve

Starts the admin UI server (FastAPI + uvicorn).

taskq ui serve [OPTIONS]

Options:

Option Type Env var fallback Default Description
--pg-dsn str TASKQ_PG_DSN postgresql://taskq:taskq@localhost:5432/taskq Postgres DSN for the admin UI pool
--schema str TASKQ_SCHEMA_NAME taskq Postgres schema
--redis-url str TASKQ_REDIS_URL None Redis URL for real-time SSE progress
--host str TASKQ_ADMIN_HOST 0.0.0.0 Bind address
--port int TASKQ_ADMIN_PORT 8080 Bind port
--migrate bool TASKQ_MIGRATE_ON_START false Apply pending migrations before starting. Aborts startup if migrations fail.

Relevant environment variables:

Variable Default Description
TASKQ_ADMIN_MAX_SSE_CONNECTIONS 50 Maximum concurrent SSE connections the admin UI will serve. Controls the connection-limit semaphore.

All options fall back to the corresponding TASKQ_* environment variable when not supplied on the command line. The admin UI requires the fastapi optional dependency group:

uv add "taskq-py[fastapi]"

The server creates a small asyncpg pool (min_size=1, max_size=4) and optionally a Redis client. The admin router is mounted at /admin. See admin-ui.md for the full UI reference.

Example:

taskq ui serve --port 8001
TASKQ_ADMIN_PORT=8001 taskq ui serve

The process blocks until killed. There is no graceful-shutdown option; use a process manager or container lifecycle hook.


taskq workgroup validate

Validates a workgroup TOML configuration file without starting any workers.

taskq workgroup validate CONFIG

Arguments:

Argument Type Description
CONFIG PATH Path to the workgroup TOML configuration file.

Prints a summary of each worker's configuration. Exits 1 if the config is missing, malformed, or contains invalid values (e.g. negative poll interval, misconfigured health check thresholds).

Example:

taskq workgroup validate workgroup.toml
# config OK — 2 worker(s), actors='myapp.actors:registry'
#   api: queues=['default'] poll=0.5s concurrency=8 health=off
#   batch: queues=['email', 'report'] poll=5.0s concurrency=2 health=on

taskq workgroup start

Starts a workgroup supervisor that manages multiple worker subprocesses from a TOML configuration file.

taskq workgroup start CONFIG

Arguments:

Argument Type Description
CONFIG PATH Path to the workgroup TOML configuration file.

The supervisor spawns one taskq worker subprocess per [[workers]] entry, manages their lifecycle, and cleanly shuts them down on SIGTERM/SIGINT. See workgroups.md for the full configuration reference and operational guidance.

Example config:

actors = "myapp.actors:registry"

[[workers]]
name = "api"
queues = ["default"]
max_concurrency = 8

[[workers]]
name = "batch"
queues = ["email", "report"]
poll_interval = 5.0
max_concurrency = 2

Example:

taskq workgroup start workgroup.toml

Exit codes:

Code Condition
0 Clean shutdown after SIGTERM or SIGINT
1 Config file not found, invalid config, or health-check PG pool initialization failure

Exit codes

Code Meaning
0 Success
1 Any failure: bad arguments, import errors, config drift, PG connection failures, health probe negative result
2 Watchdog force-exit (wedged worker) — not returned by taskq worker itself; emitted by os._exit(EXIT_WATCHDOG)
3 taskq worker --until-idle: some jobs failed during the drain
4 taskq worker --until-idle: idle-max-runtime exceeded before drain completed

The taskq worker command exits with the code returned by worker_main(). On clean SIGTERM, worker_main() returns 0. With --until-idle, worker_main() returns 0 (all jobs succeeded), 3 (some jobs failed), or 4 (idle-max-runtime exceeded). A third SIGTERM calls sys.exit(1).

The taskq workgroup start command exits 0 on clean shutdown. It exits 1 if the config file is missing, invalid, or if the optional health-check PG pool fails to initialise.