Export runs and their curves as tables an analysis reads

`fluksio export metrics` is the long table — a row per run, metric and step —
and `fluksio export runs` the wide one, a row per run with the inputs that
*vary* across the selection as columns beside its final numbers, status,
duration and the commit and digest of the code it ran. Both carry the run id
on every row, which is the join back to the run page and what makes an
exported file auditable. `Client.export_metrics`/`export_runs` answer the same
rows to a notebook.

The engine streams csv or jsonl from two routes declared above `/{run_id}`;
parquet is a client-side conversion behind the new `fluksio[parquet]` extra,
so nobody pays for pyarrow who does not want dtypes kept. The long export
reads each run through `_series`, so a cached node's curve comes with it, and
`--stride` thins each series rather than the concatenation of all of them.

Two things they needed on the way: `GET /runs` takes `?since=` and `?before=`,
so a long history pages by the last row's own timestamp instead of an offset
that shifts under it; and a read that reaches no engine now says so in half a
second rather than seven, because `runs`, `flavors`, `export` and an unwatched
`status` pass `retries=0`. Everything that submits keeps them.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01A9Hdrmf2cwNABCnE5x9UJa

backend/fluksio/api/routes/runs.py

@@ -5,11 +5,17 @@ in hours, so nothing here waits for one. The way to follow a run is to poll it
 or to listen on the flow socket, which carries its start and finish.
 """
 
+import csv
+import io
+import json
+from collections.abc import Iterator
 from datetime import UTC, datetime
+from itertools import groupby
 from typing import Any, Literal
 
 from fastapi import APIRouter, Depends, HTTPException, Request
 from fastapi.concurrency import run_in_threadpool
+from fastapi.responses import StreamingResponse
 from pydantic import BaseModel, Field, model_validator
 from sqlalchemy import func
 from sqlalchemy import select as sa_select
@@ -35,6 +41,11 @@ def elapsed_ms(since: datetime) -> float:
     return round((datetime.now(UTC) - start).total_seconds() * 1000, 2)
 
 
+def _aware(when: datetime) -> datetime:
+    """A bound as the columns store it. A naive one is read as UTC."""
+    return when if when.tzinfo else when.replace(tzinfo=UTC)
+
+
 #: Where a caller may say a run came from. "sweep" is not here because the
 #: sweep route writes it itself, and neither is a value a client made up: the
 #: column is only worth a table row if it means the same thing every time.
@@ -252,10 +263,19 @@ def read_runs(
     status: str | None = None,
     group: str | None = None,
     digest: str | None = None,
+    since: datetime | None = None,
+    before: datetime | None = None,
     limit: int = 50,
     offset: int = 0,
 ) -> Any:
-    """Runs, newest first. The queryable table an experiment log needs."""
+    """Runs, newest first. The queryable table an experiment log needs.
+
+    ``before`` is the cursor a long history is paged by: rows are newest
+    first, so handing back the last row's ``created_at`` reads the next page
+    whatever landed meanwhile — which ``offset`` cannot, since a run submitted
+    between two pages shifts every row down one. ``since`` bounds the other
+    end and is inclusive.
+    """
     statement = select(Run).order_by(col(Run.created_at).desc())
     if flow:
         statement = statement.where(col(Run.flow) == flow)
@@ -265,6 +285,10 @@ def read_runs(
         statement = statement.where(col(Run.group_id) == group)
     if digest:
         statement = statement.where(col(Run.params_digest) == digest)
+    if since:
+        statement = statement.where(col(Run.created_at) >= _aware(since))
+    if before:
+        statement = statement.where(col(Run.created_at) < _aware(before))
     statement = statement.offset(max(0, offset)).limit(min(limit, 500))
     return list(session.exec(statement))
 
@@ -296,6 +320,232 @@ def read_overview(session: SessionDep) -> Any:
     return sorted(rows.values(), key=lambda row: row.last_created_at, reverse=True)
 
 
+# ---------------------------------------------------------------------------
+# Export
+#
+# Both routes are declared above `/{run_id}`, or "export" is read as the id of
+# a run nobody has. What they are for: an analysis wants a dataframe, and the
+# alternatives are a call per run or somebody reading our schema out of
+# `fluksio.db`. The two shapes below are what an analysis actually asks for.
+# ---------------------------------------------------------------------------
+
+#: What an export is written as. Parquet is a conversion the client does over
+#: jsonl, because keeping dtypes is worth a dependency only to whoever wants it.
+ExportFormat = Literal["csv", "jsonl"]
+
+#: The columns of the long table, in order. The run is on every row: it is the
+#: join back to the run page and to what the run made, and it is what makes an
+#: exported file auditable rather than loose.
+METRIC_COLUMNS = ("run", "name", "step", "ts", "value")
+
+#: A run's own columns in the wide table. Its inputs and its final numbers
+#: follow, prefixed, so an input named "status" cannot collide with the run's.
+RUN_COLUMNS = (
+    "id",
+    "flow",
+    "status",
+    "created_at",
+    "started_at",
+    "finished_at",
+    "duration_ms",
+    "seed",
+    "group_id",
+    "code_digest",
+    "origin_commit",
+)
+
+
+def _selected(
+    session: Session,
+    flow: str | None,
+    status: str | None,
+    group: str | None,
+    ids: str,
+    since: datetime | None,
+    until: datetime | None,
+) -> list[Run]:
+    """The runs an export covers, newest first — the filters the list takes."""
+    statement = select(Run).order_by(col(Run.created_at).desc())
+    if flow:
+        statement = statement.where(col(Run.flow) == flow)
+    if status:
+        statement = statement.where(col(Run.status) == status)
+    if group:
+        statement = statement.where(col(Run.group_id) == group)
+    named = [part for part in ids.split(",") if part]
+    if named:
+        statement = statement.where(col(Run.id).in_(named))
+    if since:
+        statement = statement.where(col(Run.created_at) >= _aware(since))
+    if until:
+        statement = statement.where(col(Run.created_at) < _aware(until))
+    return list(session.exec(statement))
+
+
+def _cell(value: Any) -> Any:
+    """A value as a table holds it: a scalar, or JSON when it is not one."""
+    if hasattr(value, "isoformat"):
+        return value.isoformat()
+    if value is None or isinstance(value, (str, int, float, bool)):
+        return value
+    return json.dumps(value)
+
+
+def _varying(runs: list[Run]) -> list[str]:
+    """The inputs that differ across these runs — the axis of a sweep.
+
+    What a reader comparing arms wants as columns. Under two runs nothing can
+    differ, and a table of one run with none of its inputs in it is not worth
+    reading, so all of them are kept.
+    """
+    keys = sorted({key for run in runs for key in run.params})
+    if len(runs) < 2:
+        return keys
+    return [
+        key
+        for key in keys
+        if len({json.dumps(run.params.get(key), sort_keys=True) for run in runs}) > 1
+    ]
+
+
+def _scored(runs: list[Run]) -> list[str]:
+    """A run's final numbers: every scalar its declared outputs carry."""
+    return sorted(
+        {
+            key
+            for run in runs
+            for key, value in run.result.items()
+            if isinstance(value, (int, float)) and not isinstance(value, bool)
+        }
+    )
+
+
+def _drain(buffer: io.StringIO) -> str:
+    text = buffer.getvalue()
+    buffer.seek(0)
+    buffer.truncate(0)
+    return text
+
+
+def _csv(columns: list[str], chunks: Iterator[list[dict[str, Any]]]) -> Iterator[str]:
+    """Header first, then a chunk at a time through one reused buffer."""
+    buffer = io.StringIO()
+    writer = csv.DictWriter(buffer, fieldnames=columns)
+    writer.writeheader()
+    yield _drain(buffer)
+    for chunk in chunks:
+        writer.writerows(chunk)
+        yield _drain(buffer)
+
+
+def _stream(
+    fmt: str, name: str, columns: list[str], chunks: Iterator[list[dict[str, Any]]]
+) -> StreamingResponse:
+    """The rows out, a chunk at a time rather than a list.
+
+    Streaming because the point of an export is that it is bigger than what a
+    screen reads: a sweep's curves are millions of rows.
+    """
+    if fmt == "jsonl":
+        body: Iterator[str] = (
+            "".join(json.dumps(row) + "\n" for row in chunk) for chunk in chunks
+        )
+        media = "application/x-ndjson"
+    else:
+        body = _csv(columns, chunks)
+        media = "text/csv; charset=utf-8"
+    return StreamingResponse(
+        body,
+        media_type=media,
+        headers={"Content-Disposition": f'attachment; filename="{name}.{fmt}"'},
+    )
+
+
+@router.get("/export/metrics")
+def export_metrics(
+    session: SessionDep,
+    flow: str | None = None,
+    status: str | None = None,
+    group: str | None = None,
+    ids: str = "",
+    since: datetime | None = None,
+    until: datetime | None = None,
+    name: str = "",
+    stride: int = 1,
+    format: ExportFormat = "csv",
+) -> Any:
+    """Every selected run's series as one long table: run, name, step, ts, value.
+
+    The tidy shape a plotting library takes without reshaping. ``name`` keeps
+    the metrics it lists; ``stride`` thins each curve — per series, so asking
+    for every tenth point of two metrics gives every tenth point of both.
+    """
+    runs = _selected(session, flow, status, group, ids, since, until)
+    wanted = [part for part in name.split(",") if part]
+    step = max(1, stride)
+
+    def chunks() -> Iterator[list[dict[str, Any]]]:
+        for run in runs:
+            rows: list[dict[str, Any]] = []
+            for series, points in groupby(
+                _series(session, run.id), key=lambda row: row.name
+            ):
+                if wanted and series not in wanted:
+                    continue
+                rows.extend(
+                    {
+                        "run": run.id,
+                        "name": row.name,
+                        "step": row.step,
+                        "ts": row.ts,
+                        "value": row.value,
+                    }
+                    for row in list(points)[::step]
+                )
+            yield rows
+
+    return _stream(format, "metrics", list(METRIC_COLUMNS), chunks())
+
+
+@router.get("/export/runs")
+def export_runs(
+    session: SessionDep,
+    flow: str | None = None,
+    status: str | None = None,
+    group: str | None = None,
+    ids: str = "",
+    since: datetime | None = None,
+    until: datetime | None = None,
+    params: str = "",
+    metrics: str = "",
+    format: ExportFormat = "csv",
+) -> Any:
+    """One row per run: what it was given, what it scored, what code it ran.
+
+    The arm-comparison table. Inputs are columns rather than one JSON blob —
+    by default the ones that vary across the selection, which is the sweep
+    axis; ``params`` names them instead. ``metrics`` narrows the final numbers
+    to a few of a run's declared outputs.
+    """
+    runs = _selected(session, flow, status, group, ids, since, until)
+    inputs = [part for part in params.split(",") if part] or _varying(runs)
+    scores = [part for part in metrics.split(",") if part] or _scored(runs)
+    columns = [
+        *RUN_COLUMNS,
+        *(f"param.{key}" for key in inputs),
+        *(f"metric.{key}" for key in scores),
+    ]
+
+    def chunks() -> Iterator[list[dict[str, Any]]]:
+        for run in runs:
+            row = {column: _cell(getattr(run, column)) for column in RUN_COLUMNS}
+            row.update((f"param.{k}", _cell(run.params.get(k))) for k in inputs)
+            row.update((f"metric.{k}", _cell(run.result.get(k))) for k in scores)
+            yield [row]
+
+    return _stream(format, "runs", columns, chunks())
+
+
 @router.get("/{run_id}", response_model=RunDetail)
 def read_run(run_id: str, session: SessionDep) -> Any:
     """One run in full: what it was asked, what each node did, what it made."""

backend/fluksio/sdk/cli.py

@@ -8,6 +8,7 @@ does, imports it inside the branch that asked for it.
 from __future__ import annotations
 
 import argparse
+import csv
 import getpass
 import importlib
 import itertools
@@ -16,7 +17,7 @@ import pkgutil
 import sys
 import time
 from collections.abc import Iterator
-from contextlib import contextmanager
+from contextlib import contextmanager, nullcontext
 from pathlib import Path
 from typing import Any
 
@@ -25,6 +26,7 @@ import httpx
 from fluksio.sdk import FLOWS, Flow, SyncError
 from fluksio.sdk.client import (
     GLOBAL_DATA_DIR,
+    RETRIES,
     WAIT_TOLERANCE,
     ApiError,
     Client,
@@ -216,13 +218,18 @@ def _engine_client() -> Iterator[Client]:
 
 
 @contextmanager
-def _client_for(args: argparse.Namespace) -> Iterator[Client]:
-    """The engine this command talks to: one running somewhere, or this one."""
+def _client_for(args: argparse.Namespace, retries: int = RETRIES) -> Iterator[Client]:
+    """The engine this command talks to: one running somewhere, or this one.
+
+    ``retries=0`` is what a read passes: an engine that is not there otherwise
+    takes the backoff — seconds — to say so, and nothing in a read is worth
+    waiting out a restart for. A command that submits keeps them.
+    """
     if getattr(args, "local", False):
         with _engine_client() as client:
             yield client
     else:
-        yield Client(url=args.url, token=args.token)
+        yield Client(url=args.url, token=args.token, retries=retries)
 
 
 # ---------------------------------------------------------------------------
@@ -708,7 +715,9 @@ def cmd_status(args: argparse.Namespace) -> int:
 
     console = Console()
     try:
-        with _client_for(args) as client:
+        # A watch outlives a blip and should ride one out; a single screen is
+        # a read, and says "not answering" at once.
+        with _client_for(args, retries=RETRIES if args.watch else 0) as client:
             if not args.watch:
                 console.print(_status_screen(client))
                 return 0
@@ -751,7 +760,7 @@ def _stamp(row: dict[str, Any]) -> str:
 
 def cmd_runs(args: argparse.Namespace) -> int:
     try:
-        with _client_for(args) as client:
+        with _client_for(args, retries=0) as client:
             rows = client.runs(flow=args.flow, limit=args.limit)
     except (SyncError, ApiError) as exc:
         return _fail(str(exc))
@@ -771,7 +780,7 @@ def cmd_runs(args: argparse.Namespace) -> int:
 def cmd_flavors(args: argparse.Namespace) -> int:
     """The named sizes a node can ask for."""
     try:
-        with _client_for(args) as client:
+        with _client_for(args, retries=0) as client:
             rows = client.flavors()
     except (SyncError, ApiError) as exc:
         return _fail(str(exc))
@@ -850,6 +859,91 @@ def cmd_sweep(args: argparse.Namespace) -> int:
         return _unreachable(exc, "The runs are still on the engine; `fluksio runs`.")
 
 
+def _selection(args: argparse.Namespace) -> dict[str, Any]:
+    """The filters both exports share, minus the ones left empty."""
+    return {
+        key: value
+        for key, value in (
+            ("status", args.status),
+            ("group", args.group),
+            ("since", args.since),
+            ("until", args.until),
+        )
+        if value
+    }
+
+
+def _write_rows(rows: list[dict[str, Any]], fmt: str, out: str) -> int:
+    """The exported rows, in the format asked for, to a file or to stdout.
+
+    The engine settles the columns over the whole selection before it sends
+    anything, so every row carries the same keys in the same order and the
+    first row's are the header.
+    """
+    if not rows:
+        print("fluksio: nothing matched, so nothing was written", file=sys.stderr)
+        return 0
+    if fmt == "parquet":
+        try:
+            import pyarrow
+            import pyarrow.parquet
+        except ImportError:
+            return _fail(
+                "parquet needs pyarrow — `pip install 'fluksio[parquet]'`, or "
+                "export csv and convert it"
+            )
+        pyarrow.parquet.write_table(pyarrow.Table.from_pylist(rows), out)
+        return 0
+    with open(out, "w", newline="") if out else nullcontext(sys.stdout) as handle:
+        if fmt == "jsonl":
+            handle.writelines(json.dumps(row) + "\n" for row in rows)
+        else:
+            writer = csv.DictWriter(handle, fieldnames=list(rows[0]))
+            writer.writeheader()
+            writer.writerows(rows)
+    return 0
+
+
+def cmd_export_metrics(args: argparse.Namespace) -> int:
+    """Every selected run's numbers as one long table."""
+    if args.format == "parquet" and not args.out:
+        return _fail("--format parquet writes a file; name it with -o FILE")
+    try:
+        with _client_for(args, retries=0) as client:
+            rows = client.export_metrics(
+                flow=args.flow,
+                ids=args.run,
+                name=args.name,
+                stride=args.stride,
+                **_selection(args),
+            )
+    except (SyncError, ApiError) as exc:
+        return _fail(str(exc))
+    except httpx.HTTPError as exc:
+        return _unreachable(exc)
+    return _write_rows(rows, args.format, args.out)
+
+
+def cmd_export_runs(args: argparse.Namespace) -> int:
+    """One row per run: its inputs, its final numbers, what it ran."""
+    if args.format == "parquet" and not args.out:
+        return _fail("--format parquet writes a file; name it with -o FILE")
+    try:
+        with _client_for(args, retries=0) as client:
+            rows = client.export_runs(
+                flow=args.flow,
+                ids=args.run,
+                params=args.params,
+                metrics=args.metrics,
+                **_selection(args),
+            )
+    except (SyncError, ApiError) as exc:
+        return _fail(str(exc))
+    except httpx.HTTPError as exc:
+        return _unreachable(exc)
+    return _write_rows(rows, args.format, args.out)
+
+
 # ---------------------------------------------------------------------------
 # Wiring
 # ---------------------------------------------------------------------------
@@ -1014,3 +1108,59 @@ def add_parsers(subparsers: Any) -> None:
     )
     with_engine(parser, local=True)
     parser.set_defaults(func=cmd_sweep)
+
+    parser = subparsers.add_parser(
+        "export", help="runs and their numbers as a table an analysis reads"
+    )
+    exports = parser.add_subparsers(dest="table", required=True)
+
+    def with_selection(sub: argparse.ArgumentParser) -> None:
+        sub.add_argument("--flow", default="", help="only runs of this flow")
+        sub.add_argument(
+            "--run",
+            action="append",
+            default=[],
+            metavar="ID",
+            help="only this run; repeat for several",
+        )
+        sub.add_argument("--group", default="", help="only the runs of this sweep")
+        sub.add_argument("--status", default="", help="only runs that ended this way")
+        sub.add_argument(
+            "--since", default="", metavar="TS", help="only runs created at or after"
+        )
+        sub.add_argument(
+            "--until", default="", metavar="TS", help="only runs created before"
+        )
+        sub.add_argument("--format", default="csv", choices=("csv", "jsonl", "parquet"))
+        sub.add_argument(
+            "-o", "--out", default="", metavar="FILE", help="write here, not to stdout"
+        )
+        with_engine(sub, local=True)
+
+    sub = exports.add_parser(
+        "metrics", help="one row per run, metric and step — the tidy shape"
+    )
+    with_selection(sub)
+    sub.add_argument("--name", default="", metavar="A,B", help="only these metrics")
+    sub.add_argument(
+        "--stride",
+        type=int,
+        default=1,
+        help="keep every Nth point of each curve",
+    )
+    sub.set_defaults(func=cmd_export_metrics)
+
+    sub = exports.add_parser(
+        "runs", help="one row per run: its inputs, its final numbers, its status"
+    )
+    with_selection(sub)
+    sub.add_argument(
+        "--params",
+        default="",
+        metavar="A,B",
+        help="the inputs to put in columns (default: the ones that vary)",
+    )
+    sub.add_argument(
+        "--metrics", default="", metavar="A,B", help="the final numbers to keep"
+    )
+    sub.set_defaults(func=cmd_export_runs)

backend/fluksio/sdk/client.py

@@ -397,6 +397,75 @@ class Client:
             params={"ids": ",".join(ids), "metric": metric},
         )
 
+    def export_metrics(
+        self,
+        flow: str = "",
+        ids: Iterable[str] = (),
+        name: str = "",
+        stride: int = 1,
+        **filters: Any,
+    ) -> list[dict[str, Any]]:
+        """Runs' series as long rows: run, name, step, ts, value.
+
+        ``pd.DataFrame(client.export_metrics(flow="train"))`` is every loss
+        curve of that flow with the run id on each row. ``name`` keeps the
+        metrics it lists, ``stride`` thins each curve, and the selection is
+        narrowed the way :meth:`runs` is — ``status=``, ``group=``,
+        ``since=``, ``until=``.
+        """
+        query: dict[str, Any] = {"stride": stride}
+        if name:
+            query["name"] = name
+        return self._export("/runs/export/metrics", flow, ids, query, filters)
+
+    def export_runs(
+        self,
+        flow: str = "",
+        ids: Iterable[str] = (),
+        params: str = "",
+        metrics: str = "",
+        **filters: Any,
+    ) -> list[dict[str, Any]]:
+        """One row per run: its inputs as columns, its final numbers, its code.
+
+        The arm-comparison table. The inputs kept are the ones that vary
+        across the selection unless ``params`` names them, which is the axis a
+        sweep is read along.
+        """
+        query: dict[str, Any] = {}
+        if params:
+            query["params"] = params
+        if metrics:
+            query["metrics"] = metrics
+        return self._export("/runs/export/runs", flow, ids, query, filters)
+
+    def _export(
+        self,
+        path: str,
+        flow: str,
+        ids: Iterable[str],
+        query: dict[str, Any],
+        filters: dict[str, Any],
+    ) -> list[dict[str, Any]]:
+        """An export as parsed rows.
+
+        jsonl on the wire rather than csv: it keeps the types the engine has,
+        and it is the format the CLI converts to parquet from. Read whole —
+        the streaming is the engine's side, and a notebook wants a list.
+        """
+        if flow:
+            query["flow"] = flow
+        named = ",".join(ids)
+        if named:
+            query["ids"] = named
+        for key, value in filters.items():
+            query[key] = value.isoformat() if hasattr(value, "isoformat") else value
+        query["format"] = "jsonl"
+        response = self._request("GET", path, idempotent=True, params=query)
+        if response.status_code >= 400:
+            raise ApiError(response.status_code, _detail(response))
+        return [json.loads(line) for line in response.text.splitlines() if line]
+
     def cancel(self, run_id: str) -> Any:
         # Cancelling a cancelled run is cancelled.
         return self._call("POST", f"/runs/{run_id}/cancel", idempotent=True)