CLI

The taskq console entry point (Typer).

cli

taskq CLI entry point.

The CLI is intentionally thin today — only the commands needed to bootstrap a database. Worker and client commands will be added as those subsystems land.

Usage::

taskq migrate status
taskq migrate up [--phase pre|post] [--target VERSION] [--max-steps N]

logger module-attribute

logger: BoundLogger = structlog.get_logger('taskq.cli')

app module-attribute

app = typer.Typer(
    name="taskq",
    no_args_is_help=True,
    help="TaskQ — async Postgres-backed background jobs.",
)

migrate_app module-attribute

migrate_app = typer.Typer(
    no_args_is_help=True,
    help="Apply or inspect schema migrations.",
)

worker_app module-attribute

worker_app = typer.Typer(help='Run a TaskQ worker.')

health_app module-attribute

health_app = typer.Typer(
    no_args_is_help=True,
    help="Probe the worker's health endpoints.",
)

ui_app module-attribute

ui_app = typer.Typer(
    no_args_is_help=True, help="Admin UI server."
)

workgroup_app module-attribute

workgroup_app = typer.Typer(
    no_args_is_help=True,
    help="Manage a multi-worker process group (supervisor).",
)

actor_config_app module-attribute

actor_config_app = typer.Typer(
    no_args_is_help=True,
    help="Inspect and tune stored actor_config capacity fields on a live deployment.",
)

queues_app module-attribute

queues_app = typer.Typer(
    no_args_is_help=True,
    help="Inspect and configure queue dispatch mode and per-queue concurrency caps.",
)

worker

worker(
    actors: str = typer.Option(
        ...,
        "--actors",
        help="Module:attr reference to the actor registry (e.g. myapp.actors:registry). Resolves at startup to a Mapping[str, ActorRef] or Iterable[ActorRef].",
    ),
    force_update_actor_config: bool = typer.Option(
        False,
        "--force-update-actor-config",
        help="Allow sync_actor_config to overwrite a stored actor_config row whose queue or metadata differ from the registered values. Use for one deploy to deliberately re-route an actor, then unset. Capacity fields (max_concurrent / max_pending / result_ttl) are unaffected — use `taskq actor-config set` for those. Equivalent to env var TASKQ_FORCE_UPDATE_ACTOR_CONFIG=true.",
    ),
    queues: list[str] | None = typer.Option(
        None,
        "--queues",
        help="Queue names to consume from (repeat the flag once per queue). Overrides TASKQ_QUEUES.",
    ),
    max_concurrency: int | None = typer.Option(
        None,
        "--max-concurrency",
        help="Upper bound on concurrent jobs. Overrides TASKQ_MAX_CONCURRENCY.",
    ),
    poll_interval: float | None = typer.Option(
        None,
        "--poll-interval",
        help="Producer loop fallback polling cadence in seconds. Overrides TASKQ_POLL_INTERVAL.",
    ),
    worker_group: str | None = typer.Option(
        None,
        "--worker-group",
        help="Consumer group name for observability spans. Overrides TASKQ_WORKER_GROUP.",
    ),
    worker_label: str | None = typer.Option(
        None,
        "--worker-label",
        help="Human-readable label stored in the workers table for correlation with workgroup supervisors and external monitoring.",
    ),
    workgroup_instance: str | None = typer.Option(
        None,
        "--workgroup-instance",
        help="UUIDv7 identifying the workgroup orchestrator that launched this worker. Used for cross-process correlation and health checking.",
    ),
    health_socket_path: str | None = typer.Option(
        None,
        "--health-socket-path",
        help="Unix socket path for the health server. Overrides TASKQ_HEALTH_SOCKET_PATH. Use unique paths when running multiple workers on the same host.",
    ),
    until_idle: bool = typer.Option(
        False,
        "--until-idle",
        help="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 = typer.Option(
        None,
        "--idle-settle-window",
        help="Seconds to wait after queues appear empty before declaring drained. Overrides TASKQ_IDLE_SETTLE_WINDOW. Default 2.0. Only used with --until-idle.",
    ),
    idle_poll_interval: float | None = typer.Option(
        None,
        "--idle-poll-interval",
        help="How often to check queue depth. Overrides TASKQ_IDLE_POLL_INTERVAL. Default 1.0. Only used with --until-idle.",
    ),
    idle_max_runtime: float | None = typer.Option(
        None,
        "--idle-max-runtime",
        help="Maximum wall-clock seconds before forcing exit (code 4). Overrides TASKQ_IDLE_MAX_RUNTIME. Only used with --until-idle.",
    ),
    pg_credential_provider: str | None = typer.Option(
        None,
        "--pg-credential-provider",
        help=f"Module:attr reference to a PgCredentialProvider (e.g. {_PROVIDER_EXAMPLE}) — an instance, a zero-arg factory returning one, or the provider class. Every Postgres pool and dedicated connection is then built through it, so SIGHUP / TASKQ_RELOAD_INTERVAL rotate real credentials. Overrides TASKQ_PG_CREDENTIAL_PROVIDER (inherited by workgroup-supervised workers) via dotenvmodel.",
    ),
    redis_credential_provider: str | None = typer.Option(
        None,
        "--redis-credential-provider",
        help="Module:attr reference to a RedisCredentialProvider, in the same shapes as --pg-credential-provider. Requires TASKQ_REDIS_URL. Overrides TASKQ_REDIS_CREDENTIAL_PROVIDER via dotenvmodel.",
    ),
) -> None

Start a TaskQ worker consuming from the given actor registry.

Source code in src/taskq/cli.py

@worker_app.callback(invoke_without_command=True)
def worker(
    actors: str = typer.Option(
        ...,
        "--actors",
        help="Module:attr reference to the actor registry (e.g. myapp.actors:registry). "
        "Resolves at startup to a Mapping[str, ActorRef] or Iterable[ActorRef].",
    ),
    force_update_actor_config: bool = typer.Option(
        False,
        "--force-update-actor-config",
        help="Allow sync_actor_config to overwrite a stored actor_config row whose queue "
        "or metadata differ from the registered values. Use for one deploy to "
        "deliberately re-route an actor, then unset. Capacity fields "
        "(max_concurrent / max_pending / result_ttl) are unaffected — use "
        "`taskq actor-config set` for those. Equivalent to env var "
        "TASKQ_FORCE_UPDATE_ACTOR_CONFIG=true.",
    ),
    queues: list[str] | None = typer.Option(
        None,
        "--queues",
        help="Queue names to consume from (repeat the flag once per queue). Overrides TASKQ_QUEUES.",
    ),
    max_concurrency: int | None = typer.Option(
        None,
        "--max-concurrency",
        help="Upper bound on concurrent jobs. Overrides TASKQ_MAX_CONCURRENCY.",
    ),
    poll_interval: float | None = typer.Option(
        None,
        "--poll-interval",
        help="Producer loop fallback polling cadence in seconds. Overrides TASKQ_POLL_INTERVAL.",
    ),
    worker_group: str | None = typer.Option(
        None,
        "--worker-group",
        help="Consumer group name for observability spans. Overrides TASKQ_WORKER_GROUP.",
    ),
    worker_label: str | None = typer.Option(
        None,
        "--worker-label",
        help="Human-readable label stored in the workers table for correlation "
        "with workgroup supervisors and external monitoring.",
    ),
    workgroup_instance: str | None = typer.Option(
        None,
        "--workgroup-instance",
        help="UUIDv7 identifying the workgroup orchestrator that launched "
        "this worker. Used for cross-process correlation and health checking.",
    ),
    health_socket_path: str | None = typer.Option(
        None,
        "--health-socket-path",
        help="Unix socket path for the health server. Overrides TASKQ_HEALTH_SOCKET_PATH. "
        "Use unique paths when running multiple workers on the same host.",
    ),
    until_idle: bool = typer.Option(
        False,
        "--until-idle",
        help="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 = typer.Option(
        None,
        "--idle-settle-window",
        help="Seconds to wait after queues appear empty before declaring "
        "drained. Overrides TASKQ_IDLE_SETTLE_WINDOW. Default 2.0. "
        "Only used with --until-idle.",
    ),
    idle_poll_interval: float | None = typer.Option(
        None,
        "--idle-poll-interval",
        help="How often to check queue depth. Overrides TASKQ_IDLE_POLL_INTERVAL. "
        "Default 1.0. Only used with --until-idle.",
    ),
    idle_max_runtime: float | None = typer.Option(
        None,
        "--idle-max-runtime",
        help="Maximum wall-clock seconds before forcing exit (code 4). "
        "Overrides TASKQ_IDLE_MAX_RUNTIME. Only used with --until-idle.",
    ),
    pg_credential_provider: str | None = typer.Option(
        None,
        "--pg-credential-provider",
        help="Module:attr reference to a PgCredentialProvider (e.g. "
        f"{_PROVIDER_EXAMPLE}) — an instance, a zero-arg factory returning one, "
        "or the provider class. Every Postgres pool and dedicated connection is "
        "then built through it, so SIGHUP / TASKQ_RELOAD_INTERVAL rotate real "
        "credentials. Overrides TASKQ_PG_CREDENTIAL_PROVIDER (inherited by "
        "workgroup-supervised workers) via dotenvmodel.",
    ),
    redis_credential_provider: str | None = typer.Option(
        None,
        "--redis-credential-provider",
        help="Module:attr reference to a RedisCredentialProvider, in the same "
        "shapes as --pg-credential-provider. Requires TASKQ_REDIS_URL. "
        "Overrides TASKQ_REDIS_CREDENTIAL_PROVIDER via dotenvmodel.",
    ),
) -> None:
    """Start a TaskQ worker consuming from the given actor registry."""
    registry = _load_actor_registry(actors)

    settings = WorkerSettings.load()
    if force_update_actor_config:
        settings.force_update_actor_config = True
    if queues is not None:
        settings.queues = queues
    if max_concurrency is not None:
        settings.max_concurrency = max_concurrency
    if poll_interval is not None:
        settings.poll_interval = poll_interval
    if worker_group is not None:
        settings.worker_group = worker_group
    if worker_label is not None:
        settings.worker_label = worker_label
    if workgroup_instance is not None:
        settings.workgroup_instance = workgroup_instance
    if health_socket_path is not None:
        settings.health_socket_path = health_socket_path

    connections = _credential_connections(
        settings,
        _resolved_ref(pg_credential_provider, settings.pg_credential_provider),
        _resolved_ref(redis_credential_provider, settings.redis_credential_provider),
    )

    try:
        code = _worker_main(
            settings,
            actor_registry=registry,
            connections=connections,
            until_idle=until_idle,
            idle_settle_window=idle_settle_window,
            idle_poll_interval=idle_poll_interval,
            idle_max_runtime=idle_max_runtime,
        )
    except ActorConfigDriftList as e:
        # Why: the remedy hint is folded into ActorConfigDriftList.__str__
        # itself (see exceptions.py) — don't print it a second time here.
        typer.echo(str(e), err=True)
        raise typer.Exit(code=1) from None
    raise typer.Exit(code=code)

dev_watch

dev_watch(
    actors: Annotated[
        str,
        Argument(help="Import path: dotted.module:attr"),
    ],
    watch: Annotated[
        list[str] | None,
        Option(
            --watch,
            help="Path to watch (repeatable). Default: cwd.",
        ),
    ] = None,
    grace_period: Annotated[
        int,
        Option(
            --grace - period,
            min=0,
            help="Seconds before SIGKILL. Default: 5.",
        ),
    ] = 5,
) -> None

Run a worker in dev mode with auto-reload on file changes.

Source code in src/taskq/cli.py

@app.command("dev", help="Development utilities.")
def dev_watch(
    actors: Annotated[str, typer.Argument(help="Import path: dotted.module:attr")],
    watch: Annotated[
        list[str] | None,
        typer.Option("--watch", help="Path to watch (repeatable). Default: cwd."),
    ] = None,
    grace_period: Annotated[
        int,
        typer.Option("--grace-period", min=0, help="Seconds before SIGKILL. Default: 5."),
    ] = 5,
) -> None:
    """Run a worker in dev mode with auto-reload on file changes."""
    module_name, sep, attr_name = actors.partition(":")
    if not sep or not module_name or not attr_name:
        typer.echo(
            f"expected module:attr syntax (e.g. myapp.actors:registry); got {actors!r}",
            err=True,
        )
        raise typer.Exit(code=1)

    try:
        module = importlib.import_module(module_name)
    except ModuleNotFoundError:
        typer.echo(f"Error: cannot import '{module_name}' — module not found", err=True)
        raise typer.Exit(code=1) from None
    except Exception as exc:
        typer.echo(f"Error: cannot import '{module_name}' — {exc}", err=True)
        raise typer.Exit(code=1) from None

    try:
        getattr(module, attr_name)
    except AttributeError:
        typer.echo(
            f"Error: attribute {attr_name!r} not found in module {module_name}",
            err=True,
        )
        raise typer.Exit(code=1) from None

    watch_paths: list[str] = list(watch) if watch else [str(Path.cwd())]

    watch_display = ", ".join(str(p) for p in watch_paths)
    typer.echo(f"TaskQ dev mode — watching {watch_display}. Press Ctrl-C to stop.", err=True)

    with asyncio.Runner() as runner:
        runner.run(
            dev_watch_loop(actors, watch_paths=watch_paths, grace_period=float(grace_period))
        )

migrate_status

migrate_status(
    pg_credential_provider: str | None = typer.Option(
        None,
        "--pg-credential-provider",
        help=f"Module:attr reference to a PgCredentialProvider (e.g. {_PROVIDER_EXAMPLE}). The connection is opened through it instead of the DSN's static password. Overrides TASKQ_PG_CREDENTIAL_PROVIDER.",
    ),
) -> None

Show applied and pending migrations.

Source code in src/taskq/cli.py

@migrate_app.command("status")
def migrate_status(
    pg_credential_provider: str | None = typer.Option(
        None,
        "--pg-credential-provider",
        help="Module:attr reference to a PgCredentialProvider (e.g. "
        f"{_PROVIDER_EXAMPLE}). The connection is opened through it instead of "
        "the DSN's static password. Overrides TASKQ_PG_CREDENTIAL_PROVIDER.",
    ),
) -> None:
    """Show applied and pending migrations."""
    settings = TaskQSettings.load()
    conn_factory = _credential_conn_factory(
        str(settings.pg_dsn),
        _resolved_ref(pg_credential_provider, settings.pg_credential_provider),
        option="--pg-credential-provider",
    )
    asyncio.run(_status(settings, conn_factory=conn_factory))

migrate_up

migrate_up(
    phase: Phase | None = typer.Option(
        None, "--phase", help="Restrict to 'pre' or 'post'."
    ),
    target: str | None = typer.Option(
        None,
        "--target",
        help="Stop after this version (inclusive). E.g. 01.00.00_01",
    ),
    max_steps: int | None = typer.Option(
        None, "--max-steps", help="Cap number of applies."
    ),
    pg_credential_provider: str | None = typer.Option(
        None,
        "--pg-credential-provider",
        help=f"Module:attr reference to a PgCredentialProvider (e.g. {_PROVIDER_EXAMPLE}). The connection is opened through it instead of the DSN's static password. Overrides TASKQ_PG_CREDENTIAL_PROVIDER.",
    ),
) -> None

Apply pending migrations.

Source code in src/taskq/cli.py

@migrate_app.command("up")
def migrate_up(
    phase: migrate_mod.Phase | None = typer.Option(
        None, "--phase", help="Restrict to 'pre' or 'post'."
    ),
    target: str | None = typer.Option(
        None, "--target", help="Stop after this version (inclusive). E.g. 01.00.00_01"
    ),
    max_steps: int | None = typer.Option(None, "--max-steps", help="Cap number of applies."),
    pg_credential_provider: str | None = typer.Option(
        None,
        "--pg-credential-provider",
        help="Module:attr reference to a PgCredentialProvider (e.g. "
        f"{_PROVIDER_EXAMPLE}). The connection is opened through it instead of "
        "the DSN's static password. Overrides TASKQ_PG_CREDENTIAL_PROVIDER.",
    ),
) -> None:
    """Apply pending migrations."""
    settings = TaskQSettings.load()
    conn_factory = _credential_conn_factory(
        str(settings.pg_dsn),
        _resolved_ref(pg_credential_provider, settings.pg_credential_provider),
        option="--pg-credential-provider",
    )
    asyncio.run(
        _up(
            settings,
            phase=phase,
            target=target,
            max_steps=max_steps,
            conn_factory=conn_factory,
        )
    )

actor_config_list

actor_config_list() -> None

List every stored actor_config row.

Source code in src/taskq/cli.py

@actor_config_app.command("list")
def actor_config_list() -> None:
    """List every stored actor_config row."""
    settings = TaskQSettings.load()
    asyncio.run(_actor_config_list(settings))

actor_config_get

actor_config_get(
    actor: Annotated[str, Argument(help="Actor name.")],
) -> None

Show the stored actor_config row for one actor.

Source code in src/taskq/cli.py

@actor_config_app.command("get")
def actor_config_get(
    actor: Annotated[str, typer.Argument(help="Actor name.")],
) -> None:
    """Show the stored actor_config row for one actor."""
    settings = TaskQSettings.load()
    asyncio.run(_actor_config_get(settings, actor))

actor_config_set

actor_config_set(
    actor: Annotated[str, Argument(help="Actor name.")],
    max_concurrent: Annotated[
        int | None,
        Option(
            --max - concurrent,
            min=0,
            help="New fleet-wide concurrency cap. Takes effect on the next dispatch cycle (no worker restart) — the dispatch query re-reads this column every cycle.",
        ),
    ] = None,
    clear_max_concurrent: Annotated[
        bool,
        Option(
            --clear - max - concurrent,
            help="Set max_concurrent back to unlimited.",
        ),
    ] = False,
    max_pending: Annotated[
        int | None,
        Option(
            --max - pending,
            min=0,
            help="New queue-depth backpressure cap. Takes effect within seconds on every enqueue-side process (bounded by each client's capacity-cache TTL, default 5s) — no redeploy, no worker restart.",
        ),
    ] = None,
    clear_max_pending: Annotated[
        bool,
        Option(
            --clear - max - pending,
            help="Clear the stored override; enforcement reverts to the @actor(max_pending=...) literal.",
        ),
    ] = False,
    result_ttl: Annotated[
        float | None,
        Option(
            --result - ttl,
            min=0,
            help="New result TTL in seconds. Takes effect for jobs completing after this change (no worker restart) — the terminal-write UPDATE re-reads this column for every job.",
        ),
    ] = None,
    clear_result_ttl: Annotated[
        bool,
        Option(
            --clear - result - ttl,
            help="Set result_ttl back to unset.",
        ),
    ] = False,
) -> None

Update capacity fields on an existing actor_config row.

Only flags actually passed are changed. An actor must already have a stored row (created by a worker startup that registered it) before its capacity can be tuned here.

All three fields are live: --max-concurrent is re-read by the dispatch query every cycle and --result-ttl by the terminal-write path on every job completion (both immediate); --max-pending is re-read by every enqueue-side process through a TTL-bounded cache (default 5s staleness). No redeploy and no worker restart for any of them. Use taskq actor-config diff to see the stored value, the code literal, and which one the engine currently enforces.

Source code in src/taskq/cli.py

@actor_config_app.command("set")
def actor_config_set(
    actor: Annotated[str, typer.Argument(help="Actor name.")],
    max_concurrent: Annotated[
        int | None,
        typer.Option(
            "--max-concurrent",
            min=0,
            help="New fleet-wide concurrency cap. Takes effect on the next dispatch cycle "
            "(no worker restart) — the dispatch query re-reads this column every cycle.",
        ),
    ] = None,
    clear_max_concurrent: Annotated[
        bool, typer.Option("--clear-max-concurrent", help="Set max_concurrent back to unlimited.")
    ] = False,
    max_pending: Annotated[
        int | None,
        typer.Option(
            "--max-pending",
            min=0,
            help="New queue-depth backpressure cap. Takes effect within seconds on every "
            "enqueue-side process (bounded by each client's capacity-cache TTL, default 5s) "
            "— no redeploy, no worker restart.",
        ),
    ] = None,
    clear_max_pending: Annotated[
        bool,
        typer.Option(
            "--clear-max-pending",
            help="Clear the stored override; enforcement reverts to the @actor(max_pending=...) literal.",
        ),
    ] = False,
    result_ttl: Annotated[
        float | None,
        typer.Option(
            "--result-ttl",
            min=0,
            help="New result TTL in seconds. Takes effect for jobs completing after this "
            "change (no worker restart) — the terminal-write UPDATE re-reads this column "
            "for every job.",
        ),
    ] = None,
    clear_result_ttl: Annotated[
        bool, typer.Option("--clear-result-ttl", help="Set result_ttl back to unset.")
    ] = False,
) -> None:
    """Update capacity fields on an existing actor_config row.

    Only flags actually passed are changed. An actor must already have a
    stored row (created by a worker startup that registered it) before
    its capacity can be tuned here.

    All three fields are live: ``--max-concurrent`` is re-read by the
    dispatch query every cycle and ``--result-ttl`` by the terminal-write
    path on every job completion (both immediate); ``--max-pending`` is
    re-read by every enqueue-side process through a TTL-bounded cache
    (default 5s staleness). No redeploy and no worker restart for any of
    them. Use ``taskq actor-config diff`` to see the stored value, the
    code literal, and which one the engine currently enforces.
    """
    if max_concurrent is not None and clear_max_concurrent:
        typer.echo("--max-concurrent and --clear-max-concurrent are mutually exclusive", err=True)
        raise typer.Exit(code=1)
    if max_pending is not None and clear_max_pending:
        typer.echo("--max-pending and --clear-max-pending are mutually exclusive", err=True)
        raise typer.Exit(code=1)
    if result_ttl is not None and clear_result_ttl:
        typer.echo("--result-ttl and --clear-result-ttl are mutually exclusive", err=True)
        raise typer.Exit(code=1)

    mc: int | Unset | None = UNSET
    if clear_max_concurrent:
        mc = None
    elif max_concurrent is not None:
        mc = max_concurrent

    mp: int | Unset | None = UNSET
    if clear_max_pending:
        mp = None
    elif max_pending is not None:
        mp = max_pending

    rt: float | Unset | None = UNSET
    if clear_result_ttl:
        rt = None
    elif result_ttl is not None:
        rt = result_ttl

    if isinstance(mc, Unset) and isinstance(mp, Unset) and isinstance(rt, Unset):
        typer.echo(
            "nothing to change — pass at least one --max-concurrent/--max-pending/--result-ttl "
            "or --clear-* flag",
            err=True,
        )
        raise typer.Exit(code=1)

    settings = TaskQSettings.load()
    asyncio.run(_actor_config_set(settings, actor, mc, mp, rt))

actor_config_deregister

actor_config_deregister(
    actor: Annotated[
        str, Argument(help="Actor name to deregister.")
    ],
    force: Annotated[
        bool,
        Option(
            --force,
            help="Cancel pending/scheduled jobs and disable enabled cron schedules instead of refusing. Running jobs still block deregistration.",
        ),
    ] = False,
    purge_queue: Annotated[
        bool,
        Option(
            --purge - queue,
            help="Also delete the orphaned queues row if no other actor_config references the same queue.",
        ),
    ] = False,
) -> None

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

By default refuses if non-terminal jobs or enabled cron schedules reference the actor. Use --force to cancel pending/scheduled jobs and disable schedules. Running jobs always block (force or not). Use --purge-queue to also delete the queues row if no other actor uses it.

Exit codes: 0 success, 2 refusal (active jobs/schedules or invalid schema), 3 not found.

Source code in src/taskq/cli.py

@actor_config_app.command("deregister")
def actor_config_deregister(
    actor: Annotated[str, typer.Argument(help="Actor name to deregister.")],
    force: Annotated[
        bool,
        typer.Option(
            "--force",
            help="Cancel pending/scheduled jobs and disable enabled cron schedules"
            " instead of refusing. Running jobs still block deregistration.",
        ),
    ] = False,
    purge_queue: Annotated[
        bool,
        typer.Option(
            "--purge-queue",
            help="Also delete the orphaned queues row if no other actor_config"
            " references the same queue.",
        ),
    ] = False,
) -> None:
    """Deregister an actor: delete its actor_config row with safety checks.

    By default refuses if non-terminal jobs or enabled cron schedules
    reference the actor. Use --force to cancel pending/scheduled jobs and
    disable schedules. Running jobs always block (force or not). Use
    --purge-queue to also delete the queues row if no other actor uses it.

    Exit codes: 0 success, 2 refusal (active jobs/schedules or invalid
    schema), 3 not found.
    """
    settings = TaskQSettings.load()
    asyncio.run(_actor_config_deregister(settings, actor, force, purge_queue))

actor_config_diff

actor_config_diff(
    actors: Annotated[
        str,
        Option(
            --actors,
            help="Module:attr reference to the actor registry (e.g. myapp.actors:registry). Stored rows are compared against these code literals.",
        ),
    ],
) -> None

Diff stored actor_config rows against the code literals in a registry.

Per actor and field, shows the @actor(...) literal, the stored value, and the value the engine actually enforces right now ("effective"). Reach for this when debugging "why is my change not taking effect": a capacity literal that differs from the stored row is IGNORED at runtime — the stored value wins; tune it with taskq actor-config set — while a queue/metadata mismatch blocks the next worker startup with ActorConfigDriftList.

Source code in src/taskq/cli.py

@actor_config_app.command("diff")
def actor_config_diff(
    actors: Annotated[
        str,
        typer.Option(
            "--actors",
            help="Module:attr reference to the actor registry (e.g. myapp.actors:registry). "
            "Stored rows are compared against these code literals.",
        ),
    ],
) -> None:
    """Diff stored actor_config rows against the code literals in a registry.

    Per actor and field, shows the @actor(...) literal, the stored value,
    and the value the engine actually enforces right now ("effective").
    Reach for this when debugging "why is my change not taking effect":
    a capacity literal that differs from the stored row is IGNORED at
    runtime — the stored value wins; tune it with `taskq actor-config
    set` — while a queue/metadata mismatch blocks the next worker
    startup with ActorConfigDriftList.
    """
    registry = _load_actor_registry(actors)
    settings = TaskQSettings.load()
    asyncio.run(_actor_config_diff(settings, registry))

health_live

health_live() -> None

Source code in src/taskq/cli.py

@health_app.command("live")
def health_live() -> None:
    settings = WorkerSettings.load()
    with asyncio.Runner() as runner:
        code = runner.run(_health_request(settings, "/live"))
    raise typer.Exit(code=code)

health_ready

health_ready() -> None

Source code in src/taskq/cli.py

@health_app.command("ready")
def health_ready() -> None:
    settings = WorkerSettings.load()
    with asyncio.Runner() as runner:
        code = runner.run(_health_request(settings, "/ready"))
    raise typer.Exit(code=code)

health_metrics

health_metrics() -> None

Source code in src/taskq/cli.py

@health_app.command("metrics")
def health_metrics() -> None:
    settings = WorkerSettings.load()
    with asyncio.Runner() as runner:
        code = runner.run(_health_request(settings, "/metrics"))
    raise typer.Exit(code=code)

ui_serve

ui_serve(
    pg_dsn: str | None = typer.Option(
        None,
        "--pg-dsn",
        help="Postgres DSN. Falls back to TASKQ_PG_DSN via dotenvmodel.",
    ),
    schema: str | None = typer.Option(
        None,
        "--schema",
        help="Postgres schema name. Falls back to TASKQ_SCHEMA_NAME via dotenvmodel.",
    ),
    redis_url: str | None = typer.Option(
        None,
        "--redis-url",
        help="Redis URL for real-time mode. Falls back to TASKQ_REDIS_URL via dotenvmodel.",
    ),
    host: str | None = typer.Option(
        None,
        "--host",
        help="Bind address. Falls back to TASKQ_ADMIN_HOST via dotenvmodel.",
    ),
    port: int | None = typer.Option(
        None,
        "--port",
        help="Bind port. Falls back to TASKQ_ADMIN_PORT via dotenvmodel.",
    ),
    run_migrate: bool = typer.Option(
        False,
        "--migrate",
        help="Apply pending migrations before starting. Aborts startup if migrations fail.",
    ),
    pg_credential_provider: str | None = typer.Option(
        None,
        "--pg-credential-provider",
        help=f"Module:attr reference to a PgCredentialProvider (e.g. {_PROVIDER_EXAMPLE}). The admin pool (and --migrate) authenticate through it instead of the DSN's static password. Overrides TASKQ_PG_CREDENTIAL_PROVIDER.",
    ),
    redis_credential_provider: str | None = typer.Option(
        None,
        "--redis-credential-provider",
        help="Module:attr reference to a RedisCredentialProvider for the real-time mode client. Overrides TASKQ_REDIS_CREDENTIAL_PROVIDER.",
    ),
) -> None

Start the admin UI server on the given host:port.

Source code in src/taskq/cli.py

@ui_app.command("serve")
def ui_serve(
    pg_dsn: str | None = typer.Option(
        None,
        "--pg-dsn",
        help="Postgres DSN. Falls back to TASKQ_PG_DSN via dotenvmodel.",
    ),
    schema: str | None = typer.Option(
        None,
        "--schema",
        help="Postgres schema name. Falls back to TASKQ_SCHEMA_NAME via dotenvmodel.",
    ),
    redis_url: str | None = typer.Option(
        None,
        "--redis-url",
        help="Redis URL for real-time mode. Falls back to TASKQ_REDIS_URL via dotenvmodel.",
    ),
    host: str | None = typer.Option(
        None,  # pyright: ignore[reportArgumentType]  # Why: None signals "use settings default"; resolved below before passing to uvicorn.
        "--host",
        help="Bind address. Falls back to TASKQ_ADMIN_HOST via dotenvmodel.",
    ),
    port: int | None = typer.Option(
        None,  # pyright: ignore[reportArgumentType]  # Why: None signals "use settings default"; resolved below before passing to uvicorn.
        "--port",
        help="Bind port. Falls back to TASKQ_ADMIN_PORT via dotenvmodel.",
    ),
    run_migrate: bool = typer.Option(
        False,
        "--migrate",
        help="Apply pending migrations before starting. Aborts startup if migrations fail.",
    ),
    pg_credential_provider: str | None = typer.Option(
        None,
        "--pg-credential-provider",
        help="Module:attr reference to a PgCredentialProvider (e.g. "
        f"{_PROVIDER_EXAMPLE}). The admin pool (and --migrate) authenticate "
        "through it instead of the DSN's static password. Overrides "
        "TASKQ_PG_CREDENTIAL_PROVIDER.",
    ),
    redis_credential_provider: str | None = typer.Option(
        None,
        "--redis-credential-provider",
        help="Module:attr reference to a RedisCredentialProvider for the real-time "
        "mode client. Overrides TASKQ_REDIS_CREDENTIAL_PROVIDER.",
    ),
) -> None:
    """Start the admin UI server on the given host:port."""
    settings = TaskQSettings.load()

    resolved_dsn = pg_dsn if pg_dsn is not None else str(settings.pg_dsn)
    resolved_schema = schema if schema is not None else settings.schema_name
    resolved_redis = (
        redis_url
        if redis_url is not None
        else (str(settings.redis_url) if settings.redis_url is not None else None)
    )
    resolved_host = host if host is not None else settings.admin_host
    resolved_port = port if port is not None else settings.admin_port
    resolved_migrate = run_migrate or settings.migrate_on_start

    resolved_pg_provider_ref = _resolved_ref(
        pg_credential_provider, settings.pg_credential_provider
    )
    resolved_redis_provider_ref = _resolved_ref(
        redis_credential_provider, settings.redis_credential_provider
    )

    pool_factory: PoolFactory | None = None
    conn_factory: ConnFactory | None = None
    if resolved_pg_provider_ref is not None:
        pg_provider = _load_pg_credential_provider(
            resolved_pg_provider_ref, option="--pg-credential-provider"
        )
        pool_factory = make_pg_pool_factory(resolved_dsn, pg_provider, max_size=4)
        conn_factory = make_dedicated_conn_factory(resolved_dsn, pg_provider)

    redis_factory: RedisFactory | None = None
    if resolved_redis_provider_ref is not None:
        if resolved_redis is None:
            typer.echo(
                "--redis-credential-provider was given but no Redis URL is set - "
                "set TASKQ_REDIS_URL (or --redis-url), or drop the Redis provider.",
                err=True,
            )
            raise typer.Exit(code=1)
        redis_factory = make_redis_client_factory(
            resolved_redis,
            _load_redis_credential_provider(
                resolved_redis_provider_ref, option="--redis-credential-provider"
            ),
        )

    _ui_serve(
        resolved_dsn,
        resolved_schema,
        resolved_redis,
        resolved_host,
        resolved_port,
        resolved_migrate,
        settings,
        pool_factory=pool_factory,
        conn_factory=conn_factory,
        redis_factory=redis_factory,
    )

main

main() -> None

Console-script entry point.

Source code in src/taskq/cli.py

def main() -> None:
    """Console-script entry point."""
    app()

workgroup_start

workgroup_start(
    config: Annotated[
        Path,
        Argument(
            help="Path to the workgroup TOML configuration file."
        ),
    ],
) -> None

Start a workgroup supervisor that manages multiple worker processes.

The supervisor spawns one taskq worker subprocess per [[workers]] entry in the config file, monitors their health, restarts them on crash, and propagates shutdown signals.

Source code in src/taskq/cli.py

@workgroup_app.command("start")
def workgroup_start(
    config: Annotated[
        Path,
        typer.Argument(help="Path to the workgroup TOML configuration file."),
    ],
) -> None:
    """Start a workgroup supervisor that manages multiple worker processes.

    The supervisor spawns one ``taskq worker`` subprocess per ``[[workers]]``
    entry in the config file, monitors their health, restarts them on crash,
    and propagates shutdown signals.
    """
    if not config.exists():
        typer.echo(f"config file not found: {config}", err=True)
        raise typer.Exit(code=1)

    from taskq.worker.workgroup import run_forever

    asyncio.run(run_forever(config))

workgroup_validate

workgroup_validate(
    config: Annotated[
        Path,
        Argument(
            help="Path to the workgroup TOML configuration file."
        ),
    ],
) -> None

Validate a workgroup TOML config without starting any workers.

Source code in src/taskq/cli.py

@workgroup_app.command("validate")
def workgroup_validate(
    config: Annotated[
        Path,
        typer.Argument(help="Path to the workgroup TOML configuration file."),
    ],
) -> None:
    """Validate a workgroup TOML config without starting any workers."""
    if not config.exists():
        typer.echo(f"config file not found: {config}", err=True)
        raise typer.Exit(code=1)

    import tomllib

    from taskq.worker.workgroup import load_workgroup_config

    try:
        cfg = load_workgroup_config(config)
    except (ValueError, tomllib.TOMLDecodeError) as e:
        typer.echo(f"invalid config: {e}", err=True)
        raise typer.Exit(code=1) from None
    except OSError as e:
        typer.echo(f"failed to read config: {e}", err=True)
        raise typer.Exit(code=1) from None

    typer.echo(f"config OK — {len(cfg.workers)} worker(s), actors={cfg.actors!r}")
    for w in cfg.workers:
        health = "health=on" if w.health.enabled else "health=off"
        typer.echo(
            f"  {w.name}: queues={w.queues} "
            f"poll={w.poll_interval}s concurrency={w.max_concurrency} {health}"
        )

queues_list

queues_list() -> None

List every configured queue row.

Queues absent from this list are not missing -- queues are implicit and are created by enqueueing onto them. An absent queue runs on the defaults: strict_fifo ordering (so fairness_key has no effect) and no concurrency cap.

Source code in src/taskq/cli.py

@queues_app.command("list")
def queues_list() -> None:
    """List every configured queue row.

    Queues absent from this list are not missing -- queues are implicit and
    are created by enqueueing onto them. An absent queue runs on the
    defaults: strict_fifo ordering (so `fairness_key` has no effect) and no
    concurrency cap.
    """
    settings = TaskQSettings.load()
    asyncio.run(_queues_list(settings))

queues_get

queues_get(
    name: Annotated[str, Argument(help="Queue name.")],
) -> None

Show one queue's stored configuration.

Source code in src/taskq/cli.py

@queues_app.command("get")
def queues_get(
    name: Annotated[str, typer.Argument(help="Queue name.")],
) -> None:
    """Show one queue's stored configuration."""
    settings = TaskQSettings.load()
    asyncio.run(_queues_get(settings, name))

queues_set_mode

queues_set_mode(
    name: Annotated[str, Argument(help="Queue name.")],
    mode: Annotated[
        str,
        Argument(
            help=f"Dispatch ordering mode. One of: {join(QUEUE_MODES)}."
        ),
    ],
) -> None

Set a queue's dispatch ordering mode, creating the row if needed.

round_robin is what makes fairness_key do anything: on the default strict_fifo the key is accepted, stored, and ignored. Takes effect on the next dispatch cycle -- no worker restart.

Source code in src/taskq/cli.py

@queues_app.command("set-mode")
def queues_set_mode(
    name: Annotated[str, typer.Argument(help="Queue name.")],
    mode: Annotated[
        str,
        typer.Argument(help=f"Dispatch ordering mode. One of: {', '.join(QUEUE_MODES)}."),
    ],
) -> None:
    """Set a queue's dispatch ordering mode, creating the row if needed.

    `round_robin` is what makes `fairness_key` do anything: on the default
    `strict_fifo` the key is accepted, stored, and ignored. Takes effect on
    the next dispatch cycle -- no worker restart.
    """
    settings = TaskQSettings.load()
    asyncio.run(_queues_set_mode(settings, name, mode))

queues_set_max_concurrent

queues_set_max_concurrent(
    name: Annotated[str, Argument(help="Queue name.")],
    max_concurrent: Annotated[
        int | None,
        Option(
            --max - concurrent,
            min=1,
            help="New per-queue leased-slot cap (>= 1; pass --clear for uncapped).",
        ),
    ] = None,
    clear: Annotated[
        bool,
        Option(--clear, help="Remove the cap (unlimited)."),
    ] = False,
) -> None

Set or clear a queue's fleet-wide leased-slot concurrency cap.

Unlike actor-config set --max-concurrent, this is read once at worker startup, so it needs a worker restart to take effect. There is no 0 state: NULL (via --clear) is uncapped, and an emergency drain to 0 belongs to actor-config set --max-concurrent 0, which is per-actor.

Source code in src/taskq/cli.py

@queues_app.command("set-max-concurrent")
def queues_set_max_concurrent(
    name: Annotated[str, typer.Argument(help="Queue name.")],
    max_concurrent: Annotated[
        int | None,
        typer.Option(
            "--max-concurrent",
            min=1,
            help="New per-queue leased-slot cap (>= 1; pass --clear for uncapped).",
        ),
    ] = None,
    clear: Annotated[bool, typer.Option("--clear", help="Remove the cap (unlimited).")] = False,
) -> None:
    """Set or clear a queue's fleet-wide leased-slot concurrency cap.

    Unlike `actor-config set --max-concurrent`, this is read once at worker
    startup, so it needs a worker restart to take effect. There is no 0
    state: NULL (via --clear) is uncapped, and an emergency drain to 0
    belongs to `actor-config set --max-concurrent 0`, which is per-actor.
    """
    if clear and max_concurrent is not None:
        typer.echo("pass either --max-concurrent or --clear, not both", err=True)
        raise typer.Exit(code=1)
    if not clear and max_concurrent is None:
        typer.echo("pass --max-concurrent N or --clear", err=True)
        raise typer.Exit(code=1)
    settings = TaskQSettings.load()
    asyncio.run(_queues_set_max_concurrent(settings, name, None if clear else max_concurrent))