jf flow info and other cli updates

src/jobflow_remote/cli/flow.py

@@ -5,19 +5,21 @@
 from rich.text import Text
 
 from jobflow_remote import SETTINGS
-from jobflow_remote.cli.formatting import get_flow_info_table
+from jobflow_remote.cli.formatting import format_flow_info, get_flow_info_table
 from jobflow_remote.cli.jf import app
 from jobflow_remote.cli.jfr_typer import JFRTyper
 from jobflow_remote.cli.types import (
     days_opt,
     db_ids_opt,
     end_date_opt,
+    flow_db_id_arg,
     flow_ids_opt,
     flow_state_opt,
     force_opt,
+    job_flow_id_flag_opt,
     job_ids_opt,
-    job_state_opt,
     max_results_opt,
+    name_opt,
     reverse_sort_flag_opt,
     sort_opt,
     start_date_opt,
@@ -26,7 +28,9 @@
 from jobflow_remote.cli.utils import (
     SortOption,
     check_incompatible_opt,
+    exit_with_error_msg,
     exit_with_warning_msg,
+    get_job_db_ids,
     loading_spinner,
     out_console,
 )
@@ -97,6 +101,7 @@ def delete(
     state: flow_state_opt = None,
     start_date: start_date_opt = None,
     end_date: end_date_opt = None,
+    name: name_opt = None,
     days: days_opt = None,
     force: force_opt = False,
 ):
@@ -117,6 +122,7 @@ def delete(
             state=state,
             start_date=start_date,
             end_date=end_date,
+            name=name,
         )
 
     if not flows_info:
@@ -140,3 +146,36 @@ def delete(
     out_console.print(
         f"Deleted Flow(s) with db_id: {', '.join(str(i) for i in to_delete)}"
     )
+
+
+@app_flow.command(name="info")
+def flow_info(
+    flow_db_id: flow_db_id_arg,
+    job_id_flag: job_flow_id_flag_opt = False,
+):
+    """
+    Provide detailed information on a Flow
+    """
+    db_id, jf_id = get_job_db_ids(flow_db_id, None)
+    db_ids = job_ids = flow_ids = None
+    if db_id is not None:
+        db_ids = [db_id]
+    elif job_id_flag:
+        job_ids = [jf_id]
+    else:
+        flow_ids = [jf_id]
+
+    with loading_spinner():
+
+        jc = JobController()
+
+        flows_info = jc.get_flows_info(
+            job_ids=job_ids,
+            db_ids=db_ids,
+            flow_ids=flow_ids,
+            limit=1,
+        )
+    if not flows_info:
+        exit_with_error_msg("No data matching the request")
+
+    out_console.print(format_flow_info(flows_info[0]))

src/jobflow_remote/cli/formatting.py

@@ -141,6 +141,33 @@ def format_job_info(job_info: JobInfo, show_none: bool = False):
     return render_scope(d)
 
 
+def format_flow_info(flow_info: FlowInfo):
+
+    title = f"Flow: {flow_info.name} - {flow_info.flow_id} - {flow_info.state.name}"
+    table = Table(title=title)
+    table.title_style = "bold"
+    table.add_column("DB id")
+    table.add_column("Name")
+    table.add_column("State [Remote]")
+    table.add_column("Job id  (Index)")
+    table.add_column("Worker")
+
+    for i, job_id in enumerate(flow_info.job_ids):
+        state = flow_info.job_states[i].name
+
+        row = [
+            str(flow_info.db_ids[i]),
+            flow_info.job_names[i],
+            state,
+            f"{job_id}  ({flow_info.job_indexes[i]})",
+            flow_info.workers[i],
+        ]
+
+        table.add_row(*row)
+
+    return table
+
+
 def get_exec_config_table(exec_config: dict[str, ExecutionConfig], verbosity: int = 0):
     table = Table(title="Execution config", show_lines=verbosity > 0)
     table.add_column("Name")

src/jobflow_remote/cli/job.py

@@ -13,12 +13,15 @@
     days_opt,
     db_ids_opt,
     end_date_opt,
+    flow_ids_opt,
     job_db_id_arg,
     job_ids_indexes_opt,
     job_index_arg,
     job_state_opt,
     locked_opt,
     max_results_opt,
+    metadata_opt,
+    name_opt,
     query_opt,
     remote_state_arg,
     remote_state_opt,
@@ -30,6 +33,7 @@
 from jobflow_remote.cli.utils import (
     SortOption,
     check_incompatible_opt,
+    convert_metadata,
     exit_with_error_msg,
     exit_with_warning_msg,
     get_job_db_ids,
@@ -53,10 +57,13 @@
 def jobs_list(
     job_id: job_ids_indexes_opt = None,
     db_id: db_ids_opt = None,
+    flow_id: flow_ids_opt = None,
     state: job_state_opt = None,
     remote_state: remote_state_opt = None,
     start_date: start_date_opt = None,
     end_date: end_date_opt = None,
+    name: name_opt = None,
+    metadata: metadata_opt = None,
     days: days_opt = None,
     verbosity: verbosity_opt = 0,
     max_results: max_results_opt = 100,
@@ -71,6 +78,7 @@ def jobs_list(
     check_incompatible_opt({"state": state, "remote-state": remote_state})
     check_incompatible_opt({"start_date": start_date, "days": days})
     check_incompatible_opt({"end_date": end_date, "days": days})
+    metadata_dict = convert_metadata(metadata)
 
     job_ids_indexes = get_job_ids_indexes(job_id)
 
@@ -92,11 +100,14 @@ def jobs_list(
             jobs_info = jc.get_jobs_info(
                 job_ids=job_ids_indexes,
                 db_ids=db_id,
+                flow_ids=flow_id,
                 state=state,
                 remote_state=remote_state,
                 start_date=start_date,
                 locked=locked,
                 end_date=end_date,
+                name=name,
+                metadata=metadata_dict,
                 limit=max_results,
                 sort=sort,
             )

src/jobflow_remote/cli/types.py

@@ -82,6 +82,28 @@
     ),
 ]
 
+name_opt = Annotated[
+    Optional[str],
+    typer.Option(
+        "--name",
+        "-n",
+        help="The name. Default is an exact match, but all conventions from python fnmatch can be used (e.g. *test*)",
+    ),
+]
+
+
+metadata_opt = Annotated[
+    Optional[str],
+    typer.Option(
+        "--metadata",
+        "-meta",
+        help="A string representing the metadata to be queried. Can be either"
+        " a single key=value pair or a string with the JSON representation "
+        "of a dictionary containing the mongoDB query for the metadata "
+        'subdocument (e.g \'{"key1.key2": 1, "key3": "test"}\')',
+    ),
+]
+
 
 remote_state_arg = Annotated[
     RemoteState, typer.Argument(help="One of the remote states")
@@ -176,7 +198,7 @@
 job_db_id_arg = Annotated[
     str,
     typer.Argument(
-        help="The ID of the job can the db id (i.e. an integer) or a string (i.e. the uuid)",
+        help="The ID of the job. Can be the db id (i.e. an integer) or a string (i.e. the uuid)",
         metavar="ID",
     ),
 ]
@@ -189,6 +211,15 @@
 ]
 
 
+flow_db_id_arg = Annotated[
+    str,
+    typer.Argument(
+        help="The ID of the flow. Can the db id (i.e. an integer) or a string (i.e. the uuid)",
+        metavar="ID",
+    ),
+]
+
+
 force_opt = Annotated[
     bool,
     typer.Option(
@@ -199,6 +230,16 @@
 ]
 
 
+job_flow_id_flag_opt = Annotated[
+    bool,
+    typer.Option(
+        "--job",
+        "-j",
+        help="The passed ID will be the ID of one of the jobs"
+        " belonging to the flow, instead of the ID of the flow.",
+    ),
+]
+
 locked_opt = Annotated[
     bool,
     typer.Option(

src/jobflow_remote/cli/utils.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import functools
+import json
 import uuid
 from contextlib import contextmanager
 from enum import Enum
@@ -184,3 +185,19 @@ def check_valid_uuid(uuid_str):
         pass
 
     raise typer.BadParameter(f"UUID {uuid_str} is in the wrong format.")
+
+
+def convert_metadata(string_metadata: str | None) -> dict | None:
+    if not string_metadata:
+        return None
+
+    try:
+        metadata = json.loads(string_metadata)
+    except json.JSONDecodeError:
+        split = string_metadata.split("=")
+        if len(split) != 2:
+            raise typer.BadParameter(f"Wrong format for metadata {string_metadata}")
+
+        metadata = {split[0]: split[1]}
+
+    return metadata

src/jobflow_remote/fireworks/launchpad.py

@@ -21,6 +21,7 @@
 logger = logging.getLogger(__name__)
 
 
+FW_JOB_PATH = "spec._tasks.job"
 FW_UUID_PATH = "spec._tasks.job.uuid"
 FW_INDEX_PATH = "spec._tasks.job.index"
 REMOTE_DOC_PATH = "spec.remote"