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,100 @@
     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]}]"
+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": ("Run time", 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,
+    header_keys: list[str] | None = None,
+    stored_data_keys: list[str] | None = None,
+) -> Table:
+    stored_data_keys = stored_data_keys or []
+    all_header_keys = list(header_name_data_getter_map)
+    if not header_keys:
+        header_keys = all_header_keys[:6]
+        if verbosity >= 1:
+            header_keys += all_header_keys[6:10]
+        if verbosity == 1:
+            header_keys.append(all_header_keys[10])
+        if verbosity >= 2:
+            header_keys += all_header_keys[11:13]
 
     table = Table(title="Jobs info")
-    table.add_column("DB id")
-    table.add_column("Name")
-    table.add_column("State")
-    table.add_column("Job id  (Index)")
+    for key in header_keys:
+        table.add_column(header_name_data_getter_map[key][0])
+    for key in stored_data_keys:
+        table.add_column(key)
 
-    table.add_column("Worker")
-    table.add_column("Last updated" + time_zone_str)
-
-    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")
-
-    if verbosity >= 2:
-        table.add_column("Lock id")
-        table.add_column("Lock time" + time_zone_str)
+    default_stored_data_value = Text.from_markup("[italic grey58]none[/]")
 
     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(
+            *(
+                [header_name_data_getter_map[key][1](ji) for key in header_keys]
+                + [
+                    ji.stored_data.get(key, default_stored_data_value)
+                    if ji.stored_data
+                    else default_stored_data_value
+                    for key in stored_data_keys
+                ]
             )
-
-        table.add_row(*row)
+        )
 
     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,9 +106,25 @@ 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,
+    header_keys: Annotated[
+        Optional[list[str]],
+        typer.Option(
+            "--header-keys",
+            "-hk",
+            help="Table columns to be shown. Overrides verbosity option. Can also be set in the config file",
+        ),
+    ] = 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})
@@ -128,6 +145,11 @@ def jobs_list(
             worker_name,
         ],
     )
+    header_keys = header_keys or SETTINGS.cli_job_list_columns
+    if not set(header_keys).issubset(header_name_data_getter_map):
+        exit_with_error_msg(
+            f"Header keys not supported: {set(header_keys).difference(header_name_data_getter_map)}"
+        )
 
     job_ids_indexes = get_job_ids_indexes(job_id)
 
@@ -163,7 +185,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,
+            header_keys=header_keys,
+            stored_data_keys=stored_data_keys,
+        )
 
     out_console.print(table)
     if SETTINGS.cli_suggestions:

src/jobflow_remote/config/settings.py

@@ -30,6 +30,10 @@ class JobflowRemoteSettings(BaseSettings):
     cli_log_level: LogLevel = Field(
         LogLevel.WARN, description="The level set for logging in the CLI"
     )
+    cli_job_list_columns: list[str] = Field(
+        default_factory=list,
+        description="The list of columns to show in the `jf job list` command.",
+    )
 
     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,14 @@ def test_jobs_list(job_controller, two_flows_four_jobs) -> None:
         ["job", "list"], required_out=["Get more information about the errors"]
     )
 
+    outputs = ["WAITING", "READY", "none", "State", "DB id", "whatever"]
+    excluded = ["add1", "add2", "Name", "Job id"]
+    run_check_cli(
+        ["job", "list", "-hk", "state", "-hk", "db_id", "-sdk", "whatever"],
+        required_out=outputs,
+        excluded_out=excluded,
+    )
+
 
 def test_job_info(job_controller, two_flows_four_jobs) -> None:
     from jobflow_remote.testing.cli import run_check_cli