Use pydantic in CLI

alembic/versions/a3f1c7d94e02_normalise_ids_metadata_to_list.py

@@ -0,0 +1,92 @@
+"""normalise_ids_metadata_to_list
+
+Convert the ``ids`` and ``input_ids`` simulation metadata from their display-string
+form, ``"[core_profiles, equilibrium]"``, to a real list of IDS names.
+
+Revision ID: a3f1c7d94e02
+Revises: 6fb9b8fbac38
+Create Date: 2026-09-01 00:00:00.000000
+
+"""
+
+import json
+from typing import Any, Sequence, Union
+
+from sqlalchemy import text
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "a3f1c7d94e02"
+down_revision: Union[str, Sequence[str], None] = "6fb9b8fbac38"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+IDS_KEYS = ("ids", "input_ids")
+
+_SELECT = text("SELECT id, metadata FROM simulations WHERE metadata IS NOT NULL")
+_UPDATE = text("UPDATE simulations SET metadata = :metadata WHERE id = :sim_id")
+
+
+def _as_dict(value: Any) -> Any:
+    """Return the metadata column value as a dict.
+
+    SQLite hands back the raw JSON text while PostgreSQL decodes JSONB for us.
+    """
+    if isinstance(value, (bytes, bytearray, memoryview)):
+        value = bytes(value).decode("utf-8")
+    if isinstance(value, str):
+        try:
+            value = json.loads(value)
+        except ValueError:
+            return None
+    return value if isinstance(value, dict) else None
+
+
+def _split_ids(value: str) -> list:
+    text_value = value.strip()
+    if text_value.startswith("[") and text_value.endswith("]"):
+        text_value = text_value[1:-1]
+    return [name.strip() for name in text_value.split(",") if name.strip()]
+
+
+def _convert(convert_value) -> None:
+    conn = op.get_bind()
+    rows = conn.execute(_SELECT).fetchall()
+
+    for sim_id, metadata in rows:
+        meta_dict = _as_dict(metadata)
+        if not meta_dict:
+            continue
+
+        changed = False
+        for key in IDS_KEYS:
+            if key not in meta_dict:
+                continue
+            new_value = convert_value(meta_dict[key])
+            if new_value is not None and new_value != meta_dict[key]:
+                meta_dict[key] = new_value
+                changed = True
+
+        if changed:
+            conn.execute(_UPDATE, {"metadata": json.dumps(meta_dict), "sim_id": sim_id})
+
+
+def upgrade() -> None:
+    """Turn stringified IDS lists into real lists."""
+
+    def to_list(value: Any) -> Any:
+        return _split_ids(value) if isinstance(value, str) else None
+
+    _convert(to_list)
+
+
+def downgrade() -> None:
+    """Restore the display-string form of the IDS lists."""
+
+    def to_string(value: Any) -> Any:
+        if isinstance(value, list):
+            return "[{}]".format(", ".join(str(el) for el in value))
+        return None
+
+    _convert(to_string)

src/simdb/cli/commands/remote.py

@@ -11,6 +11,7 @@
 
 from simdb.cli.remote_api import RemoteAPI
 from simdb.database.models.simulation import Simulation
+from simdb.database.models.watcher import Watcher
 from simdb.notifications import Notification
 
 from . import check_meta_args, pass_config
@@ -270,6 +271,13 @@ def config_set_option(config: "Config", name: str, option: str, value: str):
     config.save()
 
 
+_NOTIFICATION_NAMES = {
+    value: notification.name
+    for notification, value in Watcher.NOTIFICATION_CHOICES.items()
+}
+"""Notification names by the single character the remote reports them as."""
+
+
 @remote.group(cls=RemoteSubGroup)
 def watcher():
     """Manage simulation watchers on REMOTE SimDB server."""
@@ -285,7 +293,10 @@ def list_watchers(api: RemoteAPI, sim_id: str):
     if watchers:
         click.echo(f"Watchers for simulation {sim_id}:")
         for watcher in watchers:
-            click.echo(watcher)
+            notification = _NOTIFICATION_NAMES.get(
+                watcher.notification, watcher.notification
+            )
+            click.echo(f"{watcher.username} <{watcher.email}> ({notification})")
     else:
         click.echo(f"no watchers found for simulation {sim_id}")
 

src/simdb/cli/commands/simulation.py

@@ -3,7 +3,7 @@
 import urllib.parse
 from itertools import chain
 from pathlib import Path
-from typing import Any, List, Optional, Tuple, Type
+from typing import Any, List, Optional, Tuple
 
 import appdirs
 import click
@@ -200,19 +200,22 @@ def simulation_ingest(config: Config, manifest_file: str, alias: str):
     click.echo("ALIAS: " + simulation.alias + "\nUUID: " + str(simulation.uuid))
 
 
-def n_required_args_adaptor(n) -> Type[click.Command]:
-    class NRequiredArgs(click.Command):
-        NArgs = n
+class OptionalRemoteCommand(click.Command):
+    """A command declared as `[REMOTE] ARG...` whose REMOTE may be left out."""
 
-        def parse_args(self, ctx, args):
-            if len(args) == self.NArgs:
-                args.insert(0, "")
-            super().parse_args(ctx, args)
+    def parse_args(self, ctx, args):
+        arguments = [p for p in self.get_params(ctx) if isinstance(p, click.Argument)]
+        if self._count_values_given(ctx, args, arguments) < len(arguments):
+            args = ["", *args]
+        super().parse_args(ctx, args)
 
-    return NRequiredArgs
+    def _count_values_given(self, ctx, args, arguments) -> int:
+        """Count how many of the ARGUMENTS the command line provides a value for."""
+        values = self.make_parser(ctx).parse_args(list(args))[0]
+        return sum(1 for argument in arguments if values.get(argument.name) is not None)
 
 
-@simulation.command("push", cls=n_required_args_adaptor(1))
+@simulation.command("push", cls=OptionalRemoteCommand)
 @pass_config
 @click.argument("remote", required=False)
 @click.argument("sim_id")
@@ -257,7 +260,7 @@ def simulation_push(
     click.echo(f"Successfully pushed simulation {simulation.uuid}")
 
 
-@simulation.command("pull", cls=n_required_args_adaptor(2))
+@simulation.command("pull", cls=OptionalRemoteCommand)
 @pass_config
 @click.argument("remote", required=False)
 @click.argument("sim_id")
@@ -380,7 +383,7 @@ def simulation_query(
     )
 
 
-@simulation.command("data", cls=n_required_args_adaptor(2))
+@simulation.command("data", cls=OptionalRemoteCommand)
 @pass_config
 @click.argument("remote", required=False)
 @click.argument("sim_id")
@@ -418,37 +421,36 @@ def simulation_data(
     except Exception as err:
         raise click.ClickException(str(err)) from err
 
-    click.echo(f"simulation : {result['simulation']}")
-    click.echo(f"path       : {result['path']}  (occurrence {result['occurrence']})")
+    click.echo(f"simulation : {result.simulation}")
+    click.echo(f"path       : {result.path}  (occurrence {result.occurrence})")
 
-    coordinates = result.get("coordinates") or []
+    field = result.field
+    coordinates = result.coordinates
     plot_coordinate = next(
         (
             coord
             for coord in coordinates
-            if isinstance(coord.get("data"), list)
-            and isinstance(result["field"].get("data"), list)
-            and len(coord["data"]) == len(result["field"]["data"])
+            if isinstance(coord.data, list)
+            and isinstance(field.data, list)
+            and len(coord.data) == len(field.data)
         ),
         None,
     )
-    field_is_1d = is_numeric_1d(result["field"].get("data"))
+    field_is_1d = is_numeric_1d(field.data)
     if field_is_1d:
-        show_quantity_textual_plot(
-            result["field"], label="field", x_quantity=plot_coordinate
-        )
+        show_quantity_textual_plot(field, label="field", x_quantity=plot_coordinate)
     else:
-        print_quantity(result["field"], label="field")
+        print_quantity(field, label="field")
 
     if config.verbose and coordinates:
         for coord in coordinates:
-            if field_is_1d and is_numeric_1d(coord.get("data")):
+            if field_is_1d and is_numeric_1d(coord.data):
                 continue
-            if isinstance(coord.get("data"), list):
-                print_quantity(coord, label=f"coord  {coord['name']}", show_stats=False)
+            if isinstance(coord.data, list):
+                print_quantity(coord, label=f"coord  {coord.name}", show_stats=False)
 
 
-@simulation.command("validate", cls=n_required_args_adaptor(1))
+@simulation.command("validate", cls=OptionalRemoteCommand)
 @pass_config
 @click.argument("remote", required=False)
 @click.argument("sim_id")

src/simdb/cli/commands/utils.py

@@ -8,6 +8,8 @@
 from rich.table import Table
 from rich.text import Text
 
+from simdb.remote.models import QuantityData, SimulationTraceData
+
 if TYPE_CHECKING:
     # Only importing these for type checking and documentation generation in order to
     # speed up runtime startup.
@@ -50,9 +52,9 @@ def is_numeric_1d(data: Any) -> bool:
     return isinstance(data, list) and bool(data) and all(_is_numeric(v) for v in data)
 
 
-def _quantity_axis_label(q: dict, fallback: str = "") -> str:
-    name = q.get("name") or fallback
-    units = q.get("units") or "-"
+def _quantity_axis_label(q: QuantityData, fallback: str = "") -> str:
+    name = q.name or fallback
+    units = q.units or "-"
     label = str(name).rsplit("/", 1)[-1] or str(name)
     return f"{label} [{units}]"
 
@@ -140,14 +142,14 @@ def _plot_panel(
 
 
 def show_quantity_textual_plot(
-    q: dict,
+    q: QuantityData,
     label: str = "",
-    x_quantity: Optional[dict] = None,
+    x_quantity: Optional[QuantityData] = None,
 ) -> None:
-    """Print line plot for a 1-D numeric QuantityData dict."""
-    name = q["name"]
-    units = q["units"] or "-"
-    data = q["data"]
+    """Print line plot for a 1-D numeric QuantityData."""
+    name = q.name
+    units = q.units or "-"
+    data = q.data
     if not is_numeric_1d(data):
         print_quantity(q, label=label)
         return
@@ -158,10 +160,10 @@ def show_quantity_textual_plot(
     xlabel = "index [-]"
     if (
         x_quantity
-        and is_numeric_1d(x_quantity.get("data"))
-        and len(x_quantity["data"]) == len(y_values)
+        and is_numeric_1d(x_quantity.data)
+        and len(x_quantity.data) == len(y_values)
     ):
-        x_values = [float(value) for value in x_quantity["data"]]
+        x_values = [float(value) for value in x_quantity.data]
         xlabel = _quantity_axis_label(x_quantity, fallback="x")
 
     title = label or name
@@ -191,11 +193,11 @@ def show_quantity_textual_plot(
     print_quantity(q, label=label)
 
 
-def print_quantity(q: dict, label: str = "", show_stats: bool = True) -> None:
-    """Print a QuantityData dict with array display and stats."""
-    name = q["name"]
-    units = q["units"] or "-"
-    data = q["data"]
+def print_quantity(q: QuantityData, label: str = "", show_stats: bool = True) -> None:
+    """Print a QuantityData with array display and stats."""
+    name = q.name
+    units = q.units or "-"
+    data = q.data
     title = f"[bold]{label or name}[/bold]  [dim]\\[{units}][/dim]"
 
     if not isinstance(data, list):
@@ -241,7 +243,10 @@ def _format_meta_value(meta_value: Any, max_len: int) -> str:
     if isinstance(meta_value, list):
         values = []
         for i, v in enumerate(meta_value):
-            values.append(f"{v:.2f}")
+            if isinstance(v, bool) or not isinstance(v, (int, float)):
+                values.append(str(v))
+            else:
+                values.append(f"{v:.2f}")
             if i >= max_len - 1:
                 values.append("...")
                 break
@@ -334,44 +339,36 @@ def print_simulations(
         )
 
 
-def _print_trace_sim(trace_data: dict, indentation: int):
+def _print_trace_sim(trace_data: SimulationTraceData, indentation: int):
     spaces = " " * indentation
 
-    if "error" in trace_data:
-        error = trace_data["error"]
-        click.echo(f"{spaces}{error}")
-        return
-
-    uuid = trace_data["uuid"]
-    alias = trace_data["alias"]
-    status = trace_data.get("status", "unknown")
+    status = trace_data.status or "unknown"
 
-    click.echo(f"{spaces}Simulation: {uuid}")
-    click.echo(f"{spaces}     Alias: {alias}")
+    click.echo(f"{spaces}Simulation: {trace_data.uuid}")
+    click.echo(f"{spaces}     Alias: {trace_data.alias}")
     click.echo(f"{spaces}    Status: {status}")
-    status_on_name = status + "_on"
-    if status_on_name in trace_data:
-        status_on = trace_data[status_on_name]
+    status_on_name = status.replace(" ", "_") + "_on"
+    status_on = getattr(trace_data, status_on_name, None)
+    if status_on is not None:
         label = status_on_name.replace("_", " ").capitalize()
         click.echo(f"{spaces}{label}: {status_on}")
 
-    if "replaces" in trace_data:
-        if "replaces_reason" in trace_data:
-            replaces_reason = trace_data["replaces_reason"]
-            click.echo(f"{spaces}Replaces: (reason: {replaces_reason})")
+    if trace_data.replaces is not None:
+        if trace_data.replaces_reason is not None:
+            click.echo(f"{spaces}Replaces: (reason: {trace_data.replaces_reason})")
         else:
             click.echo(f"{spaces}Replaces:")
-        _print_trace_sim(trace_data["replaces"], indentation + 2)
+        _print_trace_sim(trace_data.replaces, indentation + 2)
 
 
-def print_trace(trace_data: dict) -> None:
+def print_trace(trace_data: Optional[SimulationTraceData]) -> None:
     """
     Print the simulation trace data to the console.
 
-    :param trace_data: A dictionary containing the simulation trace data.
+    :param trace_data: The trace data of the simulation.
     :return: None
     """
-    if not trace_data:
+    if trace_data is None:
         click.echo("No simulations trace found")
         return