Option for stored data in jf job list

src/jobflow_remote/cli/flow.py

@@ -74,7 +74,7 @@ def flows_list(
     sort: sort_opt = SortOption.UPDATED_ON,
     reverse_sort: reverse_sort_flag_opt = False,
 ) -> None:
-    """Get the list of Jobs in the database."""
+    """Get the list of Flows in the database."""
     check_incompatible_opt({"start_date": start_date, "days": days, "hours": hours})
     check_incompatible_opt({"end_date": end_date, "days": days, "hours": hours})
 

src/jobflow_remote/cli/formatting.py

@@ -26,79 +26,93 @@
     from jobflow_remote.jobs.upgrade import UpgradeAction
 
 
-def get_job_info_table(jobs_info: list[JobInfo], verbosity: int) -> Table:
-    time_zone_str = f" [{time.tzname[0]}]"
-
-    table = Table(title="Jobs info")
-    table.add_column("DB id")
-    table.add_column("Name")
-    table.add_column("State")
-    table.add_column("Job id  (Index)")
-
-    table.add_column("Worker")
-    table.add_column("Last updated" + time_zone_str)
+def format_state(ji: JobInfo) -> Text:
+    state = ji.state.name
+    if ji.state in (JobState.REMOTE_ERROR, JobState.FAILED):
+        state = f"[bold red]{state}[/]"
+    elif ji.remote.retry_time_limit is not None:
+        state = f"[bold orange3]{state}[/]"
+    return Text.from_markup(state)
+
+
+def format_run_time(ji: JobInfo) -> str:
+    prefix = ""
+    if ji.state == JobState.RUNNING:
+        run_time = ji.estimated_run_time
+        prefix = "~"
+    else:
+        run_time = ji.run_time
+    if not run_time:
+        return ""
+    m, s = divmod(run_time, 60)
+    h, m = divmod(m, 60)
+    return prefix + f"{h:g}:{m:02g}"
+
+
+time_zone_str = f" [{time.tzname[0]}]"
+header_name_data_getter_map = {
+    "db_id": ("DB id", lambda ji: str(ji.db_id)),
+    "name": ("Name", lambda ji: ji.name),
+    "state": ("State", format_state),
+    "job_id": ("Job id (Index)", lambda ji: f"{ji.uuid}  ({ji.index})"),
+    "worker": ("Worker", lambda ji: ji.worker),
+    "last_updated": (
+        "Last updated" + time_zone_str,
+        lambda ji: convert_utc_time(ji.updated_on).strftime(fmt_datetime),
+    ),
+    "queue_id": ("Queue id", lambda ji: ji.remote.process_id),
+    "run_time": (Text("Run time [h:mm]", no_wrap=True), format_run_time),
+    "retry_time": (
+        "Retry time" + time_zone_str,
+        lambda ji: convert_utc_time(ji.remote.retry_time_limit).strftime(fmt_datetime)
+        if ji.remote.retry_time_limit
+        else None,
+    ),
+    "prev_state": (
+        "Prev state",
+        lambda ji: ji.previous_state.name if ji.previous_state else None,
+    ),
+    "locked": ("Locked", lambda ji: "*" if ji.lock_id is not None else None),
+    "lock_id": ("Lock id", lambda ji: str(ji.lock_id)),
+    "lock_time": (
+        "Lock time" + time_zone_str,
+        lambda ji: convert_utc_time(ji.lock_time).strftime(fmt_datetime)
+        if ji.lock_time
+        else None,
+    ),
+}
+
+
+def get_job_info_table(
+    jobs_info: list[JobInfo],
+    verbosity: int,
+    output_keys: list[str] | None = None,
+    stored_data_keys: list[str] | None = None,
+) -> Table:
+    stored_data_keys = stored_data_keys or []
+    if not output_keys or verbosity > 0:
+        all_output_keys = list(header_name_data_getter_map)
+        output_keys = all_output_keys[:6]
+        if verbosity >= 1:
+            output_keys += all_output_keys[6:10]
+        if verbosity == 1:
+            output_keys.append(all_output_keys[10])
+        if verbosity >= 2:
+            output_keys += all_output_keys[11:13]
+    all_display_keys = output_keys + stored_data_keys
 
-    if verbosity >= 1:
-        table.add_column("Queue id")
-        table.add_column("Run time")
-        table.add_column("Retry time" + time_zone_str)
-        table.add_column("Prev state")
-        if verbosity < 2:
-            table.add_column("Locked")
+    sdk_map = {
+        k: (k, lambda x, k=k: x.stored_data.get(k) if x.stored_data else None)
+        for k in stored_data_keys
+    }
+    full_map = header_name_data_getter_map | sdk_map
 
-    if verbosity >= 2:
-        table.add_column("Lock id")
-        table.add_column("Lock time" + time_zone_str)
+    table = Table(title="Jobs info")
+    for key in all_display_keys:
+        table.add_column(full_map[key][0])
 
     for ji in jobs_info:
-        state = ji.state.name
-
-        if ji.state in (JobState.REMOTE_ERROR, JobState.FAILED):
-            state = f"[bold red]{state}[/]"
-        elif ji.remote.retry_time_limit is not None:
-            state = f"[bold orange3]{state}[/]"
-
-        row = [
-            str(ji.db_id),
-            ji.name,
-            Text.from_markup(state),
-            f"{ji.uuid}  ({ji.index})",
-            ji.worker,
-            convert_utc_time(ji.updated_on).strftime(fmt_datetime),
-        ]
-
-        if verbosity >= 1:
-            row.append(ji.remote.process_id)
-            prefix = ""
-            if ji.state == JobState.RUNNING:
-                run_time = ji.estimated_run_time
-                prefix = "~"
-            else:
-                run_time = ji.run_time
-            if run_time:
-                m, s = divmod(run_time, 60)
-                h, m = divmod(m, 60)
-                row.append(prefix + f"{h:g}:{m:02g}")
-            else:
-                row.append("")
-            row.append(
-                convert_utc_time(ji.remote.retry_time_limit).strftime(fmt_datetime)
-                if ji.remote.retry_time_limit
-                else None
-            )
-            row.append(ji.previous_state.name if ji.previous_state else None)
-            if verbosity < 2:
-                row.append("*" if ji.lock_id is not None else None)
-
-        if verbosity >= 2:
-            row.append(str(ji.lock_id))
-            row.append(
-                convert_utc_time(ji.lock_time).strftime(fmt_datetime)
-                if ji.lock_time
-                else None
-            )
-
-        table.add_row(*row)
+        table.add_row(*(full_map[key][1](ji) for key in all_display_keys))
 
     return table
 

src/jobflow_remote/cli/job.py

@@ -15,6 +15,7 @@
     format_job_info,
     get_job_info_table,
     get_job_report_components,
+    header_name_data_getter_map,
 )
 from jobflow_remote.cli.jf import app
 from jobflow_remote.cli.jfr_typer import JFRTyper
@@ -105,13 +106,32 @@ def jobs_list(
             help="Select the jobs in FAILED and REMOTE_ERROR state. Incompatible with the --state option",
         ),
     ] = False,
+    stored_data_keys: Annotated[
+        Optional[list[str]],
+        typer.Option(
+            "--stored-data-key",
+            "-sdk",
+            help="Key to be shown from the stored_data field.",
+        ),
+    ] = None,
+    cli_output_keys: Annotated[
+        Optional[str],
+        typer.Option(
+            "--output",
+            "-o",
+            help=f"Table columns to be shown. Needs to be specified as string with comma separated keys, e.g."
+            f"'state,db_id,name'. Overrides the verbosity option. Can also be set in the config file. "
+            f"Available options are: {', '.join(header_name_data_getter_map)}",
+        ),
+    ] = None,
 ):
     """
-    Get the list of Jobs in the database
+    Get the list of Jobs in the database.
     """
     check_incompatible_opt({"start_date": start_date, "days": days, "hours": hours})
     check_incompatible_opt({"end_date": end_date, "days": days, "hours": hours})
     check_incompatible_opt({"state": state, "error": error})
+    check_incompatible_opt({"output": cli_output_keys, "verbosity": verbosity})
     check_query_incompatibility(
         custom_query,
         [
@@ -128,6 +148,19 @@ def jobs_list(
             worker_name,
         ],
     )
+    output_keys = (
+        cli_output_keys.split(",")
+        if cli_output_keys
+        else SETTINGS.cli_job_list_columns or []
+    )
+    if not set(output_keys).issubset(header_name_data_getter_map):
+        exit_with_error_msg(
+            f"Header keys not supported: {set(output_keys).difference(header_name_data_getter_map)}"
+        )
+    if stored_data_keys and not set(output_keys).isdisjoint(stored_data_keys):
+        exit_with_error_msg(
+            "Specifying a stored data key which is a standard column is disallowed."
+        )
 
     job_ids_indexes = get_job_ids_indexes(job_id)
 
@@ -163,7 +196,12 @@ def jobs_list(
                 sort=db_sort,
             )
 
-        table = get_job_info_table(jobs_info, verbosity=verbosity)
+        table = get_job_info_table(
+            jobs_info,
+            verbosity=verbosity,
+            output_keys=output_keys,
+            stored_data_keys=stored_data_keys,
+        )
 
     out_console.print(table)
     if SETTINGS.cli_suggestions:

src/jobflow_remote/config/settings.py

@@ -30,6 +30,11 @@ class JobflowRemoteSettings(BaseSettings):
     cli_log_level: LogLevel = Field(
         LogLevel.WARN, description="The level set for logging in the CLI"
     )
+    cli_job_list_columns: Optional[list[str]] = Field(
+        None,
+        description="The list of columns to show in the `jf job list` command. For available "
+        "options check the corresponding help: `jf job list -h`.",
+    )
 
     model_config = SettingsConfigDict(env_prefix="jfremote_")
 

src/jobflow_remote/jobs/data.py

@@ -147,6 +147,7 @@ class JobInfo(BaseModel):
     end_time: Optional[datetime] = None
     priority: int = 0
     metadata: Optional[dict] = None
+    stored_data: Optional[dict] = None
 
     @property
     def is_locked(self) -> bool:

tests/db/cli/test_job.py

@@ -32,6 +32,48 @@ def test_jobs_list(job_controller, two_flows_four_jobs) -> None:
         ["job", "list"], required_out=["Get more information about the errors"]
     )
 
+    outputs = ["WAITING", "READY", "State", "DB id", "whatever"]
+    excluded = ["add1", "add2", "Name", "Job id"]
+    run_check_cli(
+        ["job", "list", "-o", "state,db_id", "-sdk", "whatever"],
+        required_out=outputs,
+        excluded_out=excluded,
+    )
+
+    output = "Header keys not supported: {'not_existing_key'}"
+    run_check_cli(
+        ["job", "list", "-o", "state,not_existing_key"], required_out=output, error=True
+    )
+
+    output = "Options output, verbosity are incompatible"
+    run_check_cli(
+        ["job", "list", "-o", "state,name", "-vv"], required_out=output, error=True
+    )
+
+
+def test_jobs_list_settings(job_controller, two_flows_four_jobs, monkeypatch) -> None:
+    from jobflow_remote import SETTINGS
+    from jobflow_remote.testing.cli import run_check_cli
+
+    with monkeypatch.context() as m:
+        m.setattr(SETTINGS, "cli_job_list_columns", ["state", "db_id"])
+
+        outputs = ["WAITING", "READY", "State", "DB id"]
+        excluded = ["add1", "add2", "Name", "Job id"]
+        run_check_cli(
+            ["job", "list"],
+            required_out=outputs,
+            excluded_out=excluded,
+        )
+
+        # using -v will lead to a very large table which isn't fully displayed
+        columns = ["DB", "id", "Name", "Sta", "Job", "id", "Wor", "Last", "Loc"]
+        outputs = columns + [f"add{i}" for i in range(1, 5)] + ["REA", "WAI"]
+        run_check_cli(
+            ["job", "list", "-v"],
+            required_out=outputs,
+        )
+
 
 def test_job_info(job_controller, two_flows_four_jobs) -> None:
     from jobflow_remote.testing.cli import run_check_cli