Tasks Commands

CLI commands for managing quantum computing tasks.

Tasks CLI presenter - Explicit Typer commands for task management.

get_table_schema_for_list

get_table_schema_for_list() -> TableSchema

Get table rendering schema for the list command.

:return: Table schema with column specifications for task list display.

Source code in qsh/plugins/tasks/cli.py

def get_table_schema_for_list() -> TableSchema:
    """Get table rendering schema for the list command.

    :return: Table schema with column specifications for task list display.
    """
    columns = [
        ColumnSpec(
            id="id",
            header="id",
            path="id",
            priority=1,
            min_width=36,
            max_width=None,  # Always show full UUID
        ),
        ColumnSpec(
            id="task_status",
            header="task_status",
            path="task_status",
            priority=2,
            min_width=10,
            max_width=15,
            value_styles={
                "Created": "dim",
                "PayloadProcessing": "blue",
                "Scheduled": "blue bold",
                "ExecutionStarted": "green",
                "Completed": "green bold",
                "Cancelled": "yellow bold",
                "Failed": "red bold",
                "PayloadProcessingError": "red bold",
            },
        ),
        ColumnSpec(
            id="definition_id",
            header="definition_id",
            path="definition_id",
            priority=3,
            min_width=36,
            max_width=None,
        ),
        ColumnSpec(
            id="group",
            header="group",
            path="group.name",
            priority=4,
            min_width=10,
            max_width=30,
            overflow="ellipsis",
        ),
        ColumnSpec(
            id="group_id",
            header="group_id",
            path="group.id",
            priority=5,
            min_width=36,
            max_width=None,
            wide_only=True,
        ),
        ColumnSpec(
            id="created_date",
            header="created_date",
            path="created_date",
            priority=6,
            min_width=16,
            max_width=25,
            overflow="ellipsis",
            format_type="datetime",
        ),
        ColumnSpec(
            id="modified_date",
            header="modified_date",
            path="modified_date",
            priority=7,
            min_width=16,
            max_width=25,
            overflow="ellipsis",
            format_type="datetime",
        ),
        ColumnSpec(
            id="compilation_id",
            header="compilation_id",
            path="compilation_id",
            priority=8,
            min_width=10,
            max_width=36,
            overflow="ellipsis",
            wide_only=True,
        ),
        ColumnSpec(
            id="error_reasons",
            header="error_reasons",
            path="error_reasons",
            priority=9,
            min_width=10,
            max_width=25,
            overflow="ellipsis",
            wide_only=True,
        ),
        ColumnSpec(
            id="created_by",
            header="created_by",
            path="created_by",
            priority=10,
            min_width=10,
            max_width=36,
            overflow="ellipsis",
        ),
    ]

    return TableSchema(collection_path="elements", columns=columns)

register_cli

register_cli(app: Typer, ctx_provider: Callable[[], AppContext], logger: Logger) -> None

Register tasks CLI commands with the main app.

This function creates a Typer sub-application for tasks and registers all task-related commands. It's called by the plugin loader during application initialization.

:param app: Main Typer application to attach commands to. :param ctx_provider: Callable that returns the current AppContext. :param logger: Logger instance for this plugin.

Source code in qsh/plugins/tasks/cli.py

def register_cli(
    app: typer.Typer, ctx_provider: Callable[[], AppContext], logger: logging.Logger
) -> None:
    """Register tasks CLI commands with the main app.

    This function creates a Typer sub-application for tasks and registers
    all task-related commands. It's called by the plugin loader during
    application initialization.

    :param app: Main Typer application to attach commands to.
    :param ctx_provider: Callable that returns the current AppContext.
    :param logger: Logger instance for this plugin.
    """
    resource_app = typer.Typer(help="Manage tasks.", no_args_is_help=True)

    cli_commands = StandardCliCommands(
        client_class=TasksClient,
        ctx_provider=ctx_provider,
        logger=logger,
        resource_name="task",
        table_schema_factory=get_table_schema_for_list,
    )

    list_cmd = create_group_list_command(
        ctx_provider=ctx_provider,
        client_factory=lambda ctx: TasksClient(ctx),
        logger=logger,
        resource_name="task",
        table_schema_factory=get_table_schema_for_list,
        renderer=lambda payload, output: render(payload, output),
        table_renderer=lambda payload, schema, **kwargs: render_payload_table(
            format_group_summary_table_payload(payload), schema, **kwargs
        ),
    )
    create_cmd = create_group_create_command(
        ctx_provider=ctx_provider,
        client_factory=lambda ctx: TasksClient(ctx),
        logger=logger,
        resource_name="task",
        plugin_name="tasks",
        request_model=TaskCreationRequest,
        renderer=lambda payload, output: render(payload, output),
    )
    resource_app.command("list")(list_cmd)
    resource_app.command("create")(create_cmd)

    @resource_app.command("get")
    def get_cmd(
        qpu_mode: str | None = typer.Option(None, "--qpu-mode", help=QPU_MODE_HELP),
        task_id: str = typer.Argument(..., help="Task ID"),
        raw: bool = typer.Option(False, "--raw", help=RAW_MODE_RECEIVE_HELP),
        output: str | None = typer.Option(None, "-o", "--output", help=OUTPUT_FORMAT_HELP),
    ) -> None:
        """Get details of a specific task."""
        ctx = ctx_provider()
        logger.debug(f"Getting task: qpu_mode={qpu_mode}, id={task_id}, raw={raw}")

        client = TasksClient(ctx)
        result = client.get(qpu_mode=qpu_mode, id=task_id, raw=raw)

        if not raw:
            result = result.model_dump(mode="json")

        render(result, output or ctx.effective_output_format)

    show_cmd = cli_commands.create_alias_command(get_cmd, "Show")
    submit_cmd = cli_commands.create_alias_command(create_cmd, "Submit")

    resource_app.command("show")(show_cmd)
    resource_app.command("submit")(submit_cmd)

    @resource_app.command("cancel")
    def cancel_cmd(
        qpu_mode: str | None = typer.Option(None, "--qpu-mode", help=QPU_MODE_HELP),
        task_id: str = typer.Argument(..., help="Task ID to cancel"),
    ) -> None:
        """Cancel a running task.

        This command only issues the cancel request to the API. A successful
        return indicates that the server accepted the cancellation (HTTP 202)
        and is processing it. To inspect the latest task status, use
        ``qsh tasks get <task_id>`` after this command.
        """
        ctx = ctx_provider()
        logger.debug(f"Cancelling task: qpu_mode={qpu_mode}, id={task_id}")

        client = TasksClient(ctx)
        client.cancel(qpu_mode=qpu_mode, id=task_id)

        from typer import echo

        echo(f"Cancellation request sent for task {task_id}.")

    app.add_typer(resource_app, name="tasks")
    logger.debug("Registered tasks CLI commands")