Tasks Commands

CLI commands for managing quantum computing tasks.

Tasks CLI presenter - Explicit Typer commands for task management.

This module provides the CLI presentation layer for tasks. It uses the StandardCliCommands factory to generate consistent commands with minimal boilerplate.

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="created_date",
            header="created_date",
            path="created_date",
            priority=4,
            min_width=16,
            max_width=25,
            overflow="ellipsis",
            format_type="datetime",
        ),
        ColumnSpec(
            id="updated_date",
            header="updated_date",
            path="updated_date",
            priority=5,
            min_width=16,
            max_width=25,
            overflow="ellipsis",
            format_type="datetime",
        ),
        ColumnSpec(
            id="created_by",
            header="created_by",
            path="created_by",
            priority=10,
            min_width=10,
            max_width=36,
            overflow="ellipsis",
        ),
        ColumnSpec(
            id="compilation_id",
            header="compilation_id",
            path="compilation_id",
            priority=7,
            min_width=10,
            max_width=36,
            overflow="ellipsis",
            wide_only=True,
        ),
        ColumnSpec(
            id="error_reasons",
            header="error_reasons",
            path="error_reasons",
            priority=8,
            min_width=10,
            max_width=25,
            overflow="ellipsis",
            wide_only=True,
        ),
    ]

    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)

    # Create CLI commands factory
    cli_commands = StandardCliCommands(
        client_class=TasksClient,
        ctx_provider=ctx_provider,
        logger=logger,
        resource_name="task",
        table_schema_factory=get_table_schema_for_list,
    )

    # Generate standard commands
    list_cmd = cli_commands.create_list_command()
    create_cmd = cli_commands.create_create_command(TaskCreationRequest)

    # Register standard commands
    resource_app.command("list")(list_cmd)
    resource_app.command("create")(create_cmd)

    # Custom get command with proper parameter name
    @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)

        # Convert to dict if it's a Pydantic model
        if not raw:
            result = result.model_dump(mode="json")

        render(result, output or ctx.effective_output_format)

    # Aliases
    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)

    # Register custom command: cancel
    @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)

        # Explicit acknowledgement for the user; any HTTP/API error will have
        # already raised an exception before this point.
        from typer import echo

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

    # Register the tasks sub-app with the main app
    app.add_typer(resource_app, name="tasks")
    logger.debug("Registered tasks CLI commands")