22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162 | def register_cli(
app: typer.Typer, ctx_provider: Callable[[], AppContext], logger: logging.Logger
) -> None:
"""Register results CLI commands with the main app."""
resource_app = typer.Typer(help="View task results.", no_args_is_help=True)
def _log_request(
*,
task_id: str,
qpu_mode: str | None,
raw: bool,
page: int,
size: int | None,
sort: str | None,
shots_page: int | None,
shots_size: int | None,
) -> None:
"""Log the incoming results request parameters."""
logger.debug(
"Getting results: task_id=%s qpu_mode=%s raw=%s page=%s size=%s sort=%s shots_page=%s shots_size=%s",
task_id,
qpu_mode,
raw,
page,
size,
sort,
shots_page,
shots_size,
)
def _fetch_results(
ctx: AppContext,
*,
qpu_mode: str | None,
task_id: str,
raw: bool,
page: int,
size: int | None,
sort: str | None,
shots_page: int | None,
shots_size: int | None,
) -> dict:
"""Fetch results from the appropriate backend based on fidelity flags."""
client = ResultsClient(ctx)
return client.get(
qpu_mode=qpu_mode,
id=task_id,
raw_source=raw,
page=page,
size=size,
sort=sort,
shots_page=shots_page,
shots_size=shots_size,
)
def _effective_format(output: str | None, ctx: AppContext) -> str:
"""Resolve the effective output format (explicit or contextual default)."""
return output or ctx.effective_output_format
def _render_table_result(result: dict, *, ctx: AppContext, wide: bool, full_measurements: bool) -> None:
"""Render results in table mode, including Task/Subtasks/Shots sections."""
elements = result.get("elements")
tasks = elements if isinstance(elements, list) else [result]
for i, task in enumerate(tasks):
tid = task.get("task_id")
if tid and len(tasks) > 1:
console.print(f"[bold magenta]Task {i + 1} • {tid}[/bold magenta]")
render_task_tables(task, wide=wide, wrap=ctx.wrap or wide, full_measurements=full_measurements)
def _render_json_result(result: dict, *, fmt: str, ctx: AppContext, wide: bool) -> None:
"""Render results in JSON mode (verbatim server payload)."""
render(result, fmt, wrap=ctx.wrap or wide)
@resource_app.command("get")
def get_cmd(
task_id: str = typer.Argument(..., help="Task ID"),
qpu_mode: str | None = typer.Option(None, "--qpu-mode", help=QPU_MODE_HELP),
page: int = typer.Option(0, "--page", help="Page index (0-based)"),
size: int | None = typer.Option(None, "--size", help="Page size"),
sort: str | None = typer.Option(
None,
"--sort",
help="Sort criteria (e.g., 'subtask_order,asc'). Uses the server default if omitted.",
),
shots_page: int | None = typer.Option(
None,
"--shots-page",
help="Page number for shot results pagination within each subtask (0-based)",
),
shots_size: int | None = typer.Option(
None,
"--shots-size",
help="Number of shot results to include per page within each subtask",
),
raw: bool = typer.Option(
False,
"--raw",
help="Fetch unsanitized results from the Result Manager.",
),
output: str | None = typer.Option(None, "-o", "--output", help=OUTPUT_FORMAT_HELP),
wide: bool = typer.Option(False, "--wide", help=WIDE_TABLE_HELP),
) -> None:
"""Get results for a task (sanitized by default; use --raw for raw source)."""
ctx = ctx_provider()
_log_request(
task_id=task_id,
qpu_mode=qpu_mode,
raw=raw,
page=page,
size=size,
sort=sort,
shots_page=shots_page,
shots_size=shots_size,
)
result = _fetch_results(
ctx,
qpu_mode=qpu_mode,
task_id=task_id,
raw=raw,
page=page,
size=size,
sort=sort,
shots_page=shots_page,
shots_size=shots_size,
)
fmt = _effective_format(output, ctx)
if fmt == "table":
# Measurements are rendered in full by default.
_render_table_result(result, ctx=ctx, wide=wide, full_measurements=True)
return
_render_json_result(result, fmt=fmt, ctx=ctx, wide=wide)
# Alias: show -> get
resource_app.command("show")(get_cmd)
# Register with main app
app.add_typer(resource_app, name="results")
logger.debug("Registered results CLI commands")
|