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

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 stored actor_config rows that differ from the registered values. Use for one deploy to deliberately change a stored max_concurrent / queue / metadata, then unset. Equivalent to env var TASKQ_FORCE_UPDATE_ACTOR_CONFIG=true.",
    ),
    queues: list[str] | None = typer.Option(
        None,
        "--queues",
        help="Comma-separated list of queue names to consume from. 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.",
    ),
) -> 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 stored actor_config rows that differ from "
        "the registered values. Use for one deploy to deliberately change a stored "
        "max_concurrent / queue / metadata, then unset. Equivalent to env var "
        "TASKQ_FORCE_UPDATE_ACTOR_CONFIG=true.",
    ),
    queues: list[str] | None = typer.Option(
        None,
        "--queues",
        help="Comma-separated list of queue names to consume from. 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.",
    ),
) -> None:
    """Start a TaskQ worker consuming from the given actor registry."""
    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"module not found: {module_name}", err=True)
        raise typer.Exit(code=1) from None
    except Exception as exc:
        typer.echo(f"failed to import module {module_name}: {exc}", err=True)
        raise typer.Exit(code=1) from None

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

    if isinstance(raw, Mapping):
        registry: Mapping[str, ActorRef[Any, Any]] = cast(Mapping[str, ActorRef[Any, Any]], raw)
    elif (
        not isinstance(raw, (str, bytes))
        and hasattr(raw, "__iter__")
        and all(isinstance(r, ActorRef) for r in raw)  # type: ignore[arg-type]  # Why: raw is object; pyright cannot verify iterability for the isinstance call.
    ):
        registry = {r.name: r for r in raw}  # type: ignore[union-attr]  # Why: the isinstance check ensures raw is Iterable[ActorRef]; pyright cannot narrow across the all() predicate inside elif.
    else:
        typer.echo(
            "expected Mapping[str, ActorRef] or Iterable[ActorRef] at "
            f"{actors}; got {type(raw).__name__}",
            err=True,
        )
        raise typer.Exit(code=1)

    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

    try:
        code = _worker_main(settings, actor_registry=registry)
    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() -> None

Show applied and pending migrations.

Source code in src/taskq/cli.py

@migrate_app.command("status")
def migrate_status() -> None:
    """Show applied and pending migrations."""
    settings = TaskQSettings.load()
    asyncio.run(_status(settings))

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."
    ),
) -> 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."),
) -> None:
    """Apply pending migrations."""
    settings = TaskQSettings.load()
    asyncio.run(_up(settings, phase=phase, target=target, max_steps=max_steps))

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

    _ui_serve(
        resolved_dsn,
        resolved_schema,
        resolved_redis,
        resolved_host,
        resolved_port,
        resolved_migrate,
        settings,
    )

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}"
        )