CLI

The taskq console entry point (Typer).

The directive below renders taskq.cli as a Python module: the typer.Typer sub-apps and the command callbacks behind each taskq subcommand. That is the implementation, not the command surface - flags, defaults, exit codes and example output are documented in the CLI reference, which is the operator-facing document for every subcommand.

Module reference

cli

taskq CLI entry point.

Usage::

taskq migrate status
taskq migrate up [--phase pre|post] [--target VERSION] [--max-steps N] [--ddl-lock-timeout SECS]
taskq migrate disable-hypertables
taskq worker --actors myapp.actors:registry
taskq job show JOB_ID
taskq job cancel JOB_ID [--reason TEXT]
taskq job retry JOB_ID
taskq job cancel-where [--queue NAME] [--status S]... [--dry-run]
taskq insights [wait|balance|drain|cron|all] [--window 1h|6h|24h|7d] [--actor A] [--queue NAME]

The console script puts the current working directory on sys.path (see :func:main), so module:attr options resolve application modules from the directory the operator ran the command in.

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.",
)

queue_app module-attribute

queue_app = typer.Typer(
    no_args_is_help=True, help="Queue lifecycle operations."
)

job_app module-attribute

job_app = typer.Typer(
    no_args_is_help=True,
    help="Inspect individual jobs and operate on them (cancel, retry, bulk cancel).",
)

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 metadata differs from the registered value. Use for one deploy to deliberately adopt a code-side metadata change, then unset. The queue assignment is unaffected, boots never rewrite it; move an actor with `taskq actor-config move-queue`. Capacity fields (max_concurrent / max_pending / result_ttl) are likewise 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 "
        "metadata differs from the registered value. Use for one deploy to "
        "deliberately adopt a code-side metadata change, then unset. The queue "
        "assignment is unaffected, boots never rewrite it; move an actor with "
        "`taskq actor-config move-queue`. Capacity fields (max_concurrent / "
        "max_pending / result_ttl) are likewise 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
    elif WorkerSettings.resolve_cascade_value("TASKQ_HEALTH_SOCKET_PATH") is None:
        # No explicit configuration anywhere (this option, the process
        # environment, the .env cascade): the worker binds the
        # per-worker-unique default instead of the static
        # ``/tmp/taskq_health.sock``. That static value was a global
        # collision point: two workers on one host left the second with no
        # socket at all (its ``health-server-unavailable`` warning, boot
        # continuing) while ``taskq health`` silently answered with the
        # FIRST worker's state, the cross-report that made the worker that
        # just warned look healthy. A pid-unique path makes every
        # co-located worker own its socket; an explicit value here or in
        # the cascade stays authoritative verbatim, including one spelled
        # as the old default (that operator asked for the shared path and
        # keeps its collision warning).
        settings.health_socket_path = default_worker_health_socket_path()
    connections, pg_provider = _credential_connections(
        settings,
        _resolved_ref(pg_credential_provider, settings.pg_credential_provider),
        _resolved_ref(redis_credential_provider, settings.redis_credential_provider),
    )

    # Exporters are wired here, before worker_main records anything:
    # measurements a proxy instrument takes before an SDK provider exists
    # are dropped, not replayed. Logging is configured first (the same
    # idempotent setup worker_main repeats) so the wiring's startup line
    # renders in the operator's configured format.
    setup_logging(
        level=settings.log_level,
        log_format=settings.log_format,
        events_level=settings.log_events_level,
    )
    try:
        configure_exporters(settings)
    except OtelExporterConfigurationError as exc:
        typer.echo(str(exc), err=True)
        raise typer.Exit(code=1) from None

    try:
        code = _worker_main(
            settings,
            actor_registry=registry,
            connections=connections,
            pg_credential_provider=pg_provider,
            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
    # The loop is closed (asyncio restored SIG_DFL on close), but the
    # process is not gone yet: the exit path, interpreter teardown, atexit
    # hooks and thread joins all still run, and a loaded host stretches
    # every one of them. A signal in that window kills the process with
    # -15 and ERASES the drain's verdict (the observed fleet-wide-storm
    # shape - the orchestrator's redundant stop-signal landing after one
    # pod had already drained cleanly). The escalation contract only has
    # meaning while the loop lives; once it is closed there is nothing
    # left to escalate, so SIGTERM is ignored and the exit status stays
    # the drain's. This guard lives HERE - the process is genuinely about
    # to die - and not in ``worker_main``: an ignored disposition set in a
    # library function is inherited by every process that host forks
    # afterwards (the proven -9 leak), while an entrypoint's next act is
    # its own exit.
    with contextlib.suppress(ValueError):  # Why: not the main thread -> no window to guard.
        signal.signal(signal.SIGTERM, signal.SIG_IGN)
    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."
    ),
    ddl_lock_timeout: float = typer.Option(
        migrate_mod.DEFAULT_MIGRATION_DDL_LOCK_TIMEOUT,
        "--ddl-lock-timeout",
        min=0.0,
        help="Seconds a transactional migration waits for a table lock before it fails and rolls back (SET LOCAL lock_timeout). 0 waits indefinitely, parking every statement on the table behind the queued DDL.",
    ),
    allow_checksum_drift: bool = typer.Option(
        False,
        "--allow-checksum-drift",
        help="Proceed past a drifted checksum ledger: an APPLIED migration's ledger checksum differs from the bundled file, which refuses the run by default. Verify the edit is safe first; the drift is still logged as a warning on every run and the ledger keeps the stored checksum.",
    ),
    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."),
    ddl_lock_timeout: float = typer.Option(
        migrate_mod.DEFAULT_MIGRATION_DDL_LOCK_TIMEOUT,
        "--ddl-lock-timeout",
        min=0.0,
        help="Seconds a transactional migration waits for a table lock before it "
        "fails and rolls back (SET LOCAL lock_timeout). 0 waits indefinitely, "
        "parking every statement on the table behind the queued DDL.",
    ),
    allow_checksum_drift: bool = typer.Option(
        False,
        "--allow-checksum-drift",
        help="Proceed past a drifted checksum ledger: an APPLIED migration's "
        "ledger checksum differs from the bundled file, which refuses the run "
        "by default. Verify the edit is safe first; the drift is still logged "
        "as a warning on every run and the ledger keeps the stored checksum.",
    ),
    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,
            ddl_lock_timeout=ddl_lock_timeout,
            allow_checksum_drift=allow_checksum_drift,
            conn_factory=conn_factory,
        )
    )

migrate_disable_hypertables

migrate_disable_hypertables(
    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

Convert the TimescaleDB hypertables back to plain vanilla tables.

The disable mirror of the migrate up deploy step's conversion: every registered retention/compression policy is removed, and per hypertable the rows return into a table whose shape is the bundled migrations' own output (count- and twin-verified at every hand-off — see docs/guides/timescaledb.md). Flip TASKQ_TIMESCALEDB_HYPERTABLES off FIRST: with the flag still true this refuses loudly (the library-level gate is a zero-statement no-op; the CLI does not let a mistyped invocation look like a completed disable). A crashed run converges on the re-run.

Source code in src/taskq/cli.py

@migrate_app.command("disable-hypertables")
def migrate_disable_hypertables(
    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:
    """Convert the TimescaleDB hypertables back to plain vanilla tables.

    The disable mirror of the ``migrate up`` deploy step's conversion: every
    registered retention/compression policy is removed, and per hypertable the
    rows return into a table whose shape is the bundled migrations' own output
    (count- and twin-verified at every hand-off — see
    ``docs/guides/timescaledb.md``). Flip ``TASKQ_TIMESCALEDB_HYPERTABLES``
    off FIRST: with the flag still true this refuses loudly (the library-level
    gate is a zero-statement no-op; the CLI does not let a mistyped invocation
    look like a completed disable). A crashed run converges on the re-run.
    """
    settings = TaskQSettings.load()
    if settings.timescaledb_hypertables:
        typer.echo(
            "TASKQ_TIMESCALEDB_HYPERTABLES is still true: disabling now would be a "
            "zero-statement no-op. Flip the flag off (and roll the workers, so they "
            "stop expecting the hypertable schema) before running this.",
            err=True,
        )
        raise typer.Exit(code=1)
    conn_factory = _credential_conn_factory(
        str(settings.pg_dsn),
        _resolved_ref(pg_credential_provider, settings.pg_credential_provider),
        option="--pg-credential-provider",
    )
    asyncio.run(_disable_hypertables(settings, 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_move_queue

actor_config_move_queue(
    actor: Annotated[
        str, Argument(help="Actor name to move.")
    ],
    new_queue: Annotated[
        str, Argument(help="Target queue name.")
    ],
) -> None

Move an actor to a different queue in ONE operator action.

Rewrites the stored queue assignment, carries the old queue's mode and max_concurrent to the target when the target has no row of its own, and moves the actor's pending/scheduled backlog onto the target (bounded batches, then one final transaction for the flip) so old-queue strays drain through the target's consumers. Running jobs finish on the workers that claimed them; any that re-pend instead (failure retry, crash reclaim, operator retry) keep their old queue label as an audit trail but are ROUTED at dispatch by the actor's current assignment , the tail drains through the target queue's consumers, never stranded on the retired source queue. Cron fires follow the moved assignment from the flip on.

Workers boot on either side of the matching code deploy, in any order: a stale @actor(queue=...) literal logs actor-config-queue-override and adopts the stored assignment instead of refusing boot.

Exit codes: 0 moved, 2 refusal (invalid queue name, the actor is already on that queue, or the assignment changed concurrently), 3 no stored row.

Source code in src/taskq/cli.py

@actor_config_app.command("move-queue")
def actor_config_move_queue(
    actor: Annotated[str, typer.Argument(help="Actor name to move.")],
    new_queue: Annotated[str, typer.Argument(help="Target queue name.")],
) -> None:
    """Move an actor to a different queue in ONE operator action.

    Rewrites the stored queue assignment, carries the old queue's mode and
    max_concurrent to the target when the target has no row of its own, and
    moves the actor's pending/scheduled backlog onto the target (bounded
    batches, then one final transaction for the flip) so old-queue strays
    drain through the target's consumers. Running jobs finish on the
    workers that claimed them; any that re-pend instead (failure retry,
    crash reclaim, operator retry) keep their old queue label as an audit
    trail but are ROUTED at dispatch by the actor's current assignment ,
    the tail drains through the target queue's consumers, never stranded
    on the retired source queue. Cron fires follow the moved assignment
    from the flip on.

    Workers boot on either side of the matching code deploy, in any order:
    a stale `@actor(queue=...)` literal logs `actor-config-queue-override`
    and adopts the stored assignment instead of refusing boot.

    Exit codes: 0 moved, 2 refusal (invalid queue name, the actor is
    already on that queue, or the assignment changed concurrently),
    3 no stored row.
    """
    settings = TaskQSettings.load()
    asyncio.run(_actor_config_move_queue(settings, actor, new_queue))

queue_migrate

queue_migrate(
    actor: Annotated[
        str, Argument(help="Actor name to move.")
    ],
    to: Annotated[
        str,
        Option(
            --to,
            help="Target queue name. Required, the target is never defaulted.",
        ),
    ],
) -> None

Move an actor onto a different queue, and report the residual.

Applies the coordinated writes of a move, the stored assignment and the target queues row carrying the source's mode and cap, in one transaction, so a failure leaves the deployment exactly where it started rather than half-moved. The actor's pending/scheduled backlog is rewritten onto the target first, as bounded committed batches.

The target is named by --to rather than positionally: the two arguments of a move are an actor and a queue, both plain strings, and two bare positionals are the shape an operator transposes under pressure, with the consequence that the backlog drains onto a queue that was never the target.

Reports how many pending jobs still carry the source queue label. Producers still running the old literal keep placing jobs there, and those stay served by the source queue's consumers, so that count is what tells the operator when the retired queue can stop being consumed.

Exit codes: 0 moved, 2 refusal (invalid queue name, the actor is already on that queue, the assignment changed concurrently, or a drain batch timed out and the move is re-runnable), 3 no stored row.

Source code in src/taskq/cli.py

@queue_app.command("migrate")
def queue_migrate(
    actor: Annotated[str, typer.Argument(help="Actor name to move.")],
    to: Annotated[
        str,
        typer.Option(
            "--to",
            help="Target queue name. Required, the target is never defaulted.",
        ),
    ],
) -> None:
    """Move an actor onto a different queue, and report the residual.

    Applies the coordinated writes of a move, the stored assignment and
    the target `queues` row carrying the source's mode and cap, in one
    transaction, so a failure leaves the deployment exactly where it
    started rather than half-moved. The actor's pending/scheduled backlog
    is rewritten onto the target first, as bounded committed batches.

    The target is named by `--to` rather than positionally: the two
    arguments of a move are an actor and a queue, both plain strings, and
    two bare positionals are the shape an operator transposes under
    pressure, with the consequence that the backlog drains onto a queue
    that was never the target.

    Reports how many pending jobs still carry the source queue label.
    Producers still running the old literal keep placing jobs there, and
    those stay served by the source queue's consumers, so that count is
    what tells the operator when the retired queue can stop being consumed.

    Exit codes: 0 moved, 2 refusal (invalid queue name, the actor is
    already on that queue, the assignment changed concurrently, or a drain
    batch timed out and the move is re-runnable), 3 no stored row.
    """
    settings = TaskQSettings.load()
    asyncio.run(_actor_config_move_queue(settings, actor, to))

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, and a queue mismatch means the two routing halves disagree (boot adopts the stored queue and cron fires follow it while producers enqueue by their own literal); reconcile with taskq actor-config move-queue. A metadata mismatch still refuses the next worker startup with ActorConfigDriftList.

Exit codes: 0 no gate-failing drift; 1 at least one actor fails the gate, a registry actor with no stored row (it does not dispatch until one is seeded), a queue assignment mismatch (producer literals and cron routing disagree), or a metadata mismatch (startup-blocking). Capacity-only differences never affect the exit code: stored capacity is operator-owned by design, so they are reportable drift, not gate-failing drift.

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`, and a queue mismatch means the two routing halves disagree
    (boot adopts the stored queue and cron fires follow it while producers
    enqueue by their own literal); reconcile with `taskq actor-config
    move-queue`. A metadata mismatch still refuses the next worker
    startup with ActorConfigDriftList.

    Exit codes: 0 no gate-failing drift; 1 at least one actor fails the
    gate, a registry actor with no stored row (it does not dispatch until
    one is seeded), a queue assignment mismatch (producer literals and
    cron routing disagree), or a metadata mismatch (startup-blocking).
    Capacity-only differences never affect the exit code: stored capacity
    is operator-owned by design, so they are reportable drift, not
    gate-failing drift.
    """
    registry = _load_actor_registry(actors)
    settings = TaskQSettings.load()
    asyncio.run(_actor_config_diff(settings, registry))

doctor

doctor(
    actors: Annotated[
        str,
        Option(
            --actors,
            help="Module:attr reference to the actor registry (e.g. myapp.actors:registry). Stored rows are read against these registered actors.",
        ),
    ],
    platform_grace_seconds: Annotated[
        float | None,
        Option(
            --platform - grace - seconds,
            help="The orchestrator's stop grace for this deployment (Kubernetes terminationGracePeriodSeconds, ACA/ECS stop timeout, compose stop_grace_period, systemd TimeoutStopSec). When given, doctor compares it against the worker's modelled worst-case shutdown and reports the shortfall: a platform grace below it SIGKILLs the worker mid-teardown and in-flight work re-runs.",
        ),
    ] = None,
) -> None

Report capacity and configuration conditions that fail silently.

TaskQ refuses boot only on structural stored-config drift, so a whole family of misconfigurations produces no error at all: an actor with no stored row never dispatches, a leftover queues row caps an actor nobody thinks is capped, a stored max_concurrent=0 drains an actor that looks configured, and a job already pending for an actor nothing consumes waits forever. Each one's only symptom is work that does not happen. This is the one command that names them together.

The report leads with the DETECTED storage mode (vanilla, timescale-apache, or timescale-tsl - detected from the server, not read from settings) and that mode's capability consequences in one glance.

Read-only: it issues no writing statement, so it is safe to run against production mid-incident.

Exit code: 0 when the report rendered (every condition reported here is one a worker keeps running through; a diagnostic that fails the shell gets wrapped in || true and then ignored). A connection failure prints doctor failed: <cause> plus an Action line to stderr and exits 1, migrate up's failure pattern. Gating CI on drift is taskq actor-config diff, which exits non-zero by design.

Source code in src/taskq/cli.py

@app.command("doctor")
def doctor(
    actors: Annotated[
        str,
        typer.Option(
            "--actors",
            help="Module:attr reference to the actor registry (e.g. myapp.actors:registry). "
            "Stored rows are read against these registered actors.",
        ),
    ],
    platform_grace_seconds: Annotated[
        float | None,
        typer.Option(
            "--platform-grace-seconds",
            help="The orchestrator's stop grace for this deployment (Kubernetes "
            "terminationGracePeriodSeconds, ACA/ECS stop timeout, compose "
            "stop_grace_period, systemd TimeoutStopSec). When given, doctor compares "
            "it against the worker's modelled worst-case shutdown and reports the "
            "shortfall: a platform grace below it SIGKILLs the worker mid-teardown "
            "and in-flight work re-runs.",
        ),
    ] = None,
) -> None:
    """Report capacity and configuration conditions that fail silently.

    TaskQ refuses boot only on structural stored-config drift, so a whole
    family of misconfigurations produces no error at all: an actor with no
    stored row never dispatches, a leftover `queues` row caps an actor
    nobody thinks is capped, a stored `max_concurrent=0` drains an
    actor that looks configured, and a job already pending for an actor
    nothing consumes waits forever. Each one's only symptom is work that
    does not happen. This is the one command that names them together.

    The report leads with the DETECTED storage mode (vanilla,
    timescale-apache, or timescale-tsl - detected from the server, not
    read from settings) and that mode's capability consequences in one
    glance.

    Read-only: it issues no writing statement, so it is safe to run
    against production mid-incident.

    Exit code: 0 when the report rendered (every condition reported here
    is one a worker keeps running through; a diagnostic that fails the
    shell gets wrapped in `|| true` and then ignored). A connection
    failure prints `doctor failed: <cause>` plus an Action line to
    stderr and exits 1, migrate up's failure pattern. Gating CI on
    drift is `taskq actor-config diff`, which exits non-zero by design.
    """
    registry = _load_actor_registry(actors)
    settings = WorkerSettings.load()
    try:
        asyncio.run(_doctor(settings, registry, platform_grace_seconds))
    except (SystemExit, typer.Exit):
        raise
    except Exception as exc:
        # An unreachable Postgres used to escape as a raw traceback (the
        # connect's OSError, exit 1), a shape no other CLI failure surface
        # shows. The report's own contract is a rendered answer, so the
        # connection failure gets the same treatment every other CLI
        # failure gets: migrate up's two-line pattern, the cause's
        # headline and the one action to take, no traceback.
        typer.echo(f"doctor failed: {migrate_mod._exception_headline(exc)}", err=True)  # pyright: ignore[reportPrivateUsage]  # Why: the headline rule (first line, else type name) is migrate's own; sharing the helper keeps the two surfaces from drifting.
        typer.echo("Action: fix the error and re-run `taskq doctor`.", err=True)
        raise typer.Exit(code=1) from None

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
    providers: list[object] = []
    if resolved_pg_provider_ref is not None:
        pg_provider = _load_pg_credential_provider(
            resolved_pg_provider_ref, option="--pg-credential-provider"
        )
        providers.append(pg_provider)
        # Why command_timeout: this factory builds the UI's admin pool ,
        # the factory-path twin of the create_pool bound in the lifespan,
        # or the credential-provider deployment would be the one unbounded
        # admin pool left standing. (The migrate conn_factory below is
        # deliberately NOT bounded: DDL may legitimately exceed a
        # per-query budget.)
        pool_factory = make_pg_pool_factory(
            resolved_dsn,
            pg_provider,
            max_size=4,
            command_timeout=_UI_POOL_COMMAND_TIMEOUT_SECS,
            # The cadence the lifespan's rotation loop rebuilds this pool
            # on: TASKQ_RELOAD_INTERVAL when set, else derived from the
            # lease the provider grants (see ReloadSchedule).
            reload_schedule=ReloadSchedule(configured=settings.reload_interval),
        )
        # One-shot: the migration connection is opened, used and closed at
        # startup, so it declares no long-lived schedule.
        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_provider = _load_redis_credential_provider(
            resolved_redis_provider_ref, option="--redis-credential-provider"
        )
        providers.append(redis_provider)
        redis_factory = make_redis_client_factory(resolved_redis, redis_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,
        # One instance may serve both roles (EntraIdProvider); closed once.
        providers=list(dict.fromkeys(providers)),
    )

main

main() -> None

Console-script entry point.

Source code in src/taskq/cli.py

def main() -> None:
    """Console-script entry point."""
    _ensure_cwd_on_sys_path()
    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))

queues_depth

queues_depth() -> None

Per-queue depth: pending/scheduled/running/failed counts and the age of the oldest pending job, newest activity first.

Source code in src/taskq/cli.py

@queues_app.command("depth")
def queues_depth() -> None:
    """Per-queue depth: pending/scheduled/running/failed counts and the
    age of the oldest pending job, newest activity first."""
    settings = TaskQSettings.load()
    asyncio.run(_queues_depth(settings))

insights

insights(
    surface: Annotated[
        str,
        Argument(
            help=f"Which insight to render: one of {join(_INSIGHTS_SURFACES)}. Default: all."
        ),
    ] = "all",
    window: Annotated[
        str,
        Option(
            --window,
            help=f"Trailing window for the windowed reads (wait, drain, cron). One of: {join(INSIGHTS_WINDOWS)}. A window older than the archive retention answers over the surviving chunks — the retention floor is the analytics floor.",
        ),
    ] = "1h",
    actor: Annotated[
        str | None,
        Option(
            --actor,
            help="wait only: switch the wait distribution to the per-(actor, queue) grouping and keep this actor's rows. The plain per-queue read folds every actor's attempts together; the actor view separates them.",
        ),
    ] = None,
    queue: Annotated[
        str | None,
        Option(
            --queue,
            help="Keep only this queue's rows (wait, balance, drain). Refused on cron: the fan-out ledger is per schedule and has no queue dimension.",
        ),
    ] = None,
) -> None

Render the operational-insights SQL layer as operator tables.

Read-only like taskq doctor: every statement is a SELECT over BOTH retention tiers (live jobs + jobs_archive), the command issues no write, and it exits 0 whether or not there is anything to report. Safe to run against production mid-incident.

The surfaces and their columns:

• wait — how long terminalised attempts WAITED before their claim, per queue. Columns: count (terminalised attempts in the window), p50/p95/max (the wait distribution), segment. Wait per attempt is started_at - scheduled_at (the dispatch claim latency). segment = clean for first-delivery attempts (snooze_count = 0 and rate_limit_blocked_count = 0) — the subset a queue-latency SLO is written against; segment = deferred for rows whose scheduled_at a snooze or a rate limit moved forward: RESCHEDULES ARE EXCLUDED from the clean percentiles, and a deferred row's wait measures only the FINAL leg (the deferred time is by construction not in the number). --actor switches to the per-(actor, queue) grouping and keeps one actor's rows.

• balance — is the fleet imbalanced, per queue. Columns: depth (pending rows due now — the work a worker could claim this instant), live (workers subscribed to the queue and heartbeating in the last 30s), effective_capacity (the routed actors' summed max_concurrent x live workers: the queue's in-flight ceiling under the current fleet), utilization (depth ÷ effective capacity), oldest_due (age of the oldest due job: a large depth with a tiny age is a burst, a small depth with a large age is a strand), flag. Rows with utilization > 1 are MARKED !! over threshold (more due work than one wave of capacity can absorb); !! no capacity marks the starvation shape (no live worker or no actor capacity serves the queue).

• drain — seconds-to-drain, per queue: due depth ÷ completions per second over the window. This is a THROUGHPUT extrapolation, not a promise: it assumes the next window looks like the last one, that the queue's workers stay up, and that nothing enqueues behind the current depth. A queue whose window carried no completions has NO estimate: its eta renders no traffic in window — never 0, which would read as "already drained". Widen --window (up to the archive retention floor) before trusting anything else.

• cron — the per-schedule fan-out ledger. Columns: schedule (the schedule id, shortened), actor, cron, fires (jobs the schedule enqueued in the window), cleared (of those, terminalised), outstanding (the schedule's non-terminal backlog right now, windowless), verdict. !! runaway trending marks fires > cleared in BOTH the current and the prior equal window — two consecutive windows of fan-out outrunning clearance is the runaway shape; one window is a burst. Confounds: a DST allof schedule legitimately doubles a fire in the overlap hour (read the verdict against the schedule's dst_strategy), and a budget-deferred fire enqueues NO row, so a deferral reads as zero fires here — its record is the taskq.cron.budget_deferrals counter, not this ledger.

• all (the default) — every surface, in the order above.

Source code in src/taskq/cli.py

@app.command("insights")
def insights(
    surface: Annotated[
        str,
        typer.Argument(
            help=f"Which insight to render: one of {', '.join(_INSIGHTS_SURFACES)}. Default: all."
        ),
    ] = "all",
    window: Annotated[
        str,
        typer.Option(
            "--window",
            help=f"Trailing window for the windowed reads (wait, drain, cron). One of: "
            f"{', '.join(INSIGHTS_WINDOWS)}. A window older than the archive retention "
            "answers over the surviving chunks — the retention floor is the analytics floor.",
        ),
    ] = "1h",
    actor: Annotated[
        str | None,
        typer.Option(
            "--actor",
            help="wait only: switch the wait distribution to the per-(actor, queue) grouping "
            "and keep this actor's rows. The plain per-queue read folds every actor's "
            "attempts together; the actor view separates them.",
        ),
    ] = None,
    queue: Annotated[
        str | None,
        typer.Option(
            "--queue",
            help="Keep only this queue's rows (wait, balance, drain). Refused on cron: the "
            "fan-out ledger is per schedule and has no queue dimension.",
        ),
    ] = None,
) -> None:
    """Render the operational-insights SQL layer as operator tables.

    Read-only like `taskq doctor`: every statement is a SELECT over BOTH
    retention tiers (live `jobs` + `jobs_archive`), the command issues no
    write, and it exits 0 whether or not there is anything to report.
    Safe to run against production mid-incident.

    The surfaces and their columns:

    * `wait` — how long terminalised attempts WAITED before their claim,
      per queue. Columns: `count` (terminalised attempts in the window),
      `p50`/`p95`/`max` (the wait distribution), `segment`. Wait per
      attempt is `started_at - scheduled_at` (the dispatch claim
      latency). `segment` = `clean` for first-delivery attempts
      (`snooze_count = 0` and `rate_limit_blocked_count = 0`) — the
      subset a queue-latency SLO is written against; `segment` =
      `deferred` for rows whose `scheduled_at` a snooze or a rate limit
      moved forward: RESCHEDULES ARE EXCLUDED from the clean
      percentiles, and a deferred row's wait measures only the FINAL
      leg (the deferred time is by construction not in the number).
      `--actor` switches to the per-(actor, queue) grouping and keeps
      one actor's rows.

    * `balance` — is the fleet imbalanced, per queue. Columns: `depth`
      (pending rows due now — the work a worker could claim this
      instant), `live` (workers subscribed to the queue and heartbeating
      in the last 30s), `effective_capacity` (the routed actors'
      summed `max_concurrent` x live workers: the queue's in-flight
      ceiling under the current fleet), `utilization` (depth ÷
      effective capacity), `oldest_due` (age of the oldest due job:
      a large depth with a tiny age is a burst, a small depth with a
      large age is a strand), `flag`. Rows with utilization > 1 are
      MARKED `!! over threshold` (more due work than one wave of
      capacity can absorb); `!! no capacity` marks the starvation shape
      (no live worker or no actor capacity serves the queue).

    * `drain` — seconds-to-drain, per queue: due depth ÷ completions
      per second over the window. This is a THROUGHPUT extrapolation,
      not a promise: it assumes the next window looks like the last
      one, that the queue's workers stay up, and that nothing enqueues
      behind the current depth. A queue whose window carried no
      completions has NO estimate: its eta renders `no traffic in
      window` — never 0, which would read as "already drained". Widen
      `--window` (up to the archive retention floor) before trusting
      anything else.

    * `cron` — the per-schedule fan-out ledger. Columns: `schedule`
      (the schedule id, shortened), `actor`, `cron`, `fires` (jobs the
      schedule enqueued in the window), `cleared` (of those,
      terminalised), `outstanding` (the schedule's non-terminal backlog
      right now, windowless), `verdict`. `!! runaway trending` marks
      fires > cleared in BOTH the current and the prior equal window —
      two consecutive windows of fan-out outrunning clearance is the
      runaway shape; one window is a burst. Confounds: a DST `allof`
      schedule legitimately doubles a fire in the overlap hour (read
      the verdict against the schedule's dst_strategy), and a
      budget-deferred fire enqueues NO row, so a deferral reads as zero
      fires here — its record is the `taskq.cron.budget_deferrals`
      counter, not this ledger.

    * `all` (the default) — every surface, in the order above.
    """
    if surface not in _INSIGHTS_SURFACES:
        typer.echo(
            f"unknown insights surface: {surface!r} "
            f"(expected one of: {', '.join(_INSIGHTS_SURFACES)})",
            err=True,
        )
        raise typer.Exit(code=1)
    window_delta = INSIGHTS_WINDOWS.get(window)
    if window_delta is None:
        typer.echo(
            f"invalid --window: {window!r} (expected one of: {', '.join(INSIGHTS_WINDOWS)})",
            err=True,
        )
        raise typer.Exit(code=1)
    if actor is not None and surface not in ("wait", "all"):
        typer.echo(
            "--actor applies to the wait surface only: the wait distribution is the "
            "one read with an actor dimension",
            err=True,
        )
        raise typer.Exit(code=1)
    if queue is not None and surface == "cron":
        typer.echo(
            "--queue does not apply to the cron surface: the fan-out ledger is per "
            "schedule and has no queue dimension",
            err=True,
        )
        raise typer.Exit(code=1)
    settings = TaskQSettings.load()
    asyncio.run(_insights(settings, surface, window, window_delta, actor, queue))

job_show

job_show(
    job_id: Annotated[str, Argument(help="Job id (UUID).")],
    show_traceback: Annotated[
        bool,
        Option(
            --traceback,
            help="Also print the stored error_traceback (can be large).",
        ),
    ] = False,
    show_payload: Annotated[
        bool,
        Option(
            --payload,
            help="Also print the stored payload jsonb (can be large).",
        ),
    ] = False,
) -> None

Show one job's stored row, from jobs or jobs_archive.

Source code in src/taskq/cli.py

@job_app.command("show")
def job_show(
    job_id: Annotated[str, typer.Argument(help="Job id (UUID).")],
    show_traceback: Annotated[
        bool,
        typer.Option("--traceback", help="Also print the stored error_traceback (can be large)."),
    ] = False,
    show_payload: Annotated[
        bool,
        typer.Option("--payload", help="Also print the stored payload jsonb (can be large)."),
    ] = False,
) -> None:
    """Show one job's stored row, from `jobs` or `jobs_archive`."""
    settings = TaskQSettings.load()
    asyncio.run(
        _job_show(settings, job_id, show_traceback=show_traceback, show_payload=show_payload)
    )

job_cancel

job_cancel(
    job_id: Annotated[str, Argument(help="Job id (UUID).")],
    reason: Annotated[
        str | None,
        Option(
            --reason,
            help="Recorded on the job's cancel_request event.",
        ),
    ] = None,
) -> None

Cancel one job: pending/scheduled goes straight to terminal cancelled; running gets a cooperative cancel request the worker's cancel controller acts on.

Source code in src/taskq/cli.py

@job_app.command("cancel")
def job_cancel(
    job_id: Annotated[str, typer.Argument(help="Job id (UUID).")],
    reason: Annotated[
        str | None,
        typer.Option("--reason", help="Recorded on the job's cancel_request event."),
    ] = None,
) -> None:
    """Cancel one job: pending/scheduled goes straight to terminal
    `cancelled`; running gets a cooperative cancel request the worker's
    cancel controller acts on."""
    settings = TaskQSettings.load()
    asyncio.run(_job_cancel(settings, job_id, reason))

job_retry

job_retry(
    job_id: Annotated[str, Argument(help="Job id (UUID).")],
) -> None

Re-pend a resting job (every terminal status, including succeeded and abandoned). The attempt counter is not reset.

Source code in src/taskq/cli.py

@job_app.command("retry")
def job_retry(
    job_id: Annotated[str, typer.Argument(help="Job id (UUID).")],
) -> None:
    """Re-pend a resting job (every terminal status, including `succeeded`
    and `abandoned`). The attempt counter is not reset."""
    settings = TaskQSettings.load()
    asyncio.run(_job_retry(settings, job_id))

job_cancel_where

job_cancel_where(
    queue: Annotated[
        str | None,
        Option(
            --queue, help="Cancel only jobs on this queue."
        ),
    ] = None,
    status: Annotated[
        list[str] | None,
        Option(
            --status,
            help="Cancel only jobs in this status. Repeatable. Valid: pending, scheduled, running, succeeded, failed, cancelled, crashed, abandoned.",
        ),
    ] = None,
    actor: Annotated[
        str | None,
        Option(
            --actor,
            help="Cancel only jobs for this actor name.",
        ),
    ] = None,
    tag: Annotated[
        list[str] | None,
        Option(
            --tag,
            help="Cancel only jobs carrying this tag. Repeatable.",
        ),
    ] = None,
    older_than: Annotated[
        str | None,
        Option(
            --older - than,
            help="Only jobs enqueued before now minus this duration (e.g. 45, 30m, 2h, 7d, 2w).",
        ),
    ] = None,
    reason: Annotated[
        str | None,
        Option(
            --reason,
            help="Recorded on every matched job's cancel_request event.",
        ),
    ] = None,
    dry_run: Annotated[
        bool,
        Option(
            --dry - run,
            help="Print the matching count and sample ids; write nothing.",
        ),
    ] = False,
) -> None

Bulk cancel every job matching the given filters.

Refuses to run with no filter at all: a naked full-table cancel is the one mistake this command cannot walk back.

Source code in src/taskq/cli.py

@job_app.command("cancel-where")
def job_cancel_where(
    queue: Annotated[
        str | None, typer.Option("--queue", help="Cancel only jobs on this queue.")
    ] = None,
    status: Annotated[
        list[str] | None,
        typer.Option(
            "--status",
            help="Cancel only jobs in this status. Repeatable. "
            "Valid: pending, scheduled, running, succeeded, failed, cancelled, "
            "crashed, abandoned.",
        ),
    ] = None,
    actor: Annotated[
        str | None, typer.Option("--actor", help="Cancel only jobs for this actor name.")
    ] = None,
    tag: Annotated[
        list[str] | None,
        typer.Option("--tag", help="Cancel only jobs carrying this tag. Repeatable."),
    ] = None,
    older_than: Annotated[
        str | None,
        typer.Option(
            "--older-than",
            help="Only jobs enqueued before now minus this duration (e.g. 45, 30m, 2h, 7d, 2w).",
        ),
    ] = None,
    reason: Annotated[
        str | None,
        typer.Option("--reason", help="Recorded on every matched job's cancel_request event."),
    ] = None,
    dry_run: Annotated[
        bool,
        typer.Option("--dry-run", help="Print the matching count and sample ids; write nothing."),
    ] = False,
) -> None:
    """Bulk cancel every job matching the given filters.

    Refuses to run with no filter at all: a naked full-table cancel is
    the one mistake this command cannot walk back.
    """
    settings = TaskQSettings.load()
    asyncio.run(
        _job_cancel_where(
            settings,
            queue=queue,
            status=tuple(status) if status else None,
            actor=actor,
            tags=tuple(tag) if tag else None,
            older_than=older_than,
            reason=reason,
            dry_run=dry_run,
        )
    )

job_events

job_events(
    job_id: Annotated[str, Argument(help="Job id (UUID).")],
) -> None

List one job's job_events timeline (kind, occurred_at, detail).

Source code in src/taskq/cli.py

@job_app.command("events")
def job_events(
    job_id: Annotated[str, typer.Argument(help="Job id (UUID).")],
) -> None:
    """List one job's job_events timeline (kind, occurred_at, detail)."""
    settings = TaskQSettings.load()
    asyncio.run(_job_events(settings, job_id))