Runs export: a truncated one counts the selection into X-Total-Count; the SDK and fluksio export warn how many runs were left out
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WgBfkm4rCmtd1T4vHWjEPx
This commit is contained in:
@@ -512,15 +512,22 @@ def _selected(
|
||||
ids: str,
|
||||
since: datetime | None,
|
||||
until: datetime | None,
|
||||
) -> list[Run]:
|
||||
"""The runs an export covers, newest first — the filters the list takes.
|
||||
) -> tuple[list[Run], int]:
|
||||
"""The runs an export covers, newest first, and how many the filters match.
|
||||
|
||||
Capped: the filters bound a *sensible* request and nothing bounded an
|
||||
unfiltered one, so asking for everything read every row of the table into
|
||||
memory before a byte was sent. A truncated export says so in a header.
|
||||
memory before a byte was sent. A truncated export says so in a header, and
|
||||
only then is the selection counted, since otherwise the rows are the count.
|
||||
"""
|
||||
statement = _selection(flow, status, group, ids, since, until)
|
||||
return list(session.exec(statement.limit(EXPORT_CAP + 1)))
|
||||
runs = list(session.exec(statement.limit(EXPORT_CAP + 1)))
|
||||
if len(runs) <= EXPORT_CAP:
|
||||
return runs, len(runs)
|
||||
total = session.exec(
|
||||
select(func.count()).select_from(statement.order_by(None).subquery())
|
||||
).one()
|
||||
return runs[:EXPORT_CAP], total
|
||||
|
||||
|
||||
def _cell(value: Any) -> Any:
|
||||
@@ -613,7 +620,7 @@ def _stream(
|
||||
name: str,
|
||||
columns: list[str],
|
||||
chunks: Iterator[list[dict[str, Any]]],
|
||||
truncated: bool = False,
|
||||
total: int = 0,
|
||||
) -> StreamingResponse:
|
||||
"""The rows out, a chunk at a time rather than a list.
|
||||
|
||||
@@ -629,11 +636,12 @@ def _stream(
|
||||
body = _csv(columns, chunks)
|
||||
media = "text/csv; charset=utf-8"
|
||||
headers = {"Content-Disposition": f'attachment; filename="{name}.{fmt}"'}
|
||||
if truncated:
|
||||
if total > EXPORT_CAP:
|
||||
# The body is otherwise indistinguishable from a complete one, and an
|
||||
# export quietly missing its oldest runs is the wrong thing to hand
|
||||
# somebody who is about to plot it.
|
||||
# somebody who is about to plot it. The total says how many it missed.
|
||||
headers["X-Truncated"] = str(EXPORT_CAP)
|
||||
headers["X-Total-Count"] = str(total)
|
||||
return StreamingResponse(body, media_type=media, headers=headers)
|
||||
|
||||
|
||||
@@ -656,9 +664,7 @@ def export_metrics(
|
||||
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)
|
||||
truncated = len(runs) > EXPORT_CAP
|
||||
runs = runs[:EXPORT_CAP]
|
||||
runs, total = _selected(session, flow, status, group, ids, since, until)
|
||||
wanted = [part for part in name.split(",") if part]
|
||||
step = max(1, stride)
|
||||
|
||||
@@ -682,7 +688,7 @@ def export_metrics(
|
||||
)
|
||||
yield rows
|
||||
|
||||
return _stream(format, "metrics", list(METRIC_COLUMNS), chunks(), truncated)
|
||||
return _stream(format, "metrics", list(METRIC_COLUMNS), chunks(), total)
|
||||
|
||||
|
||||
@router.get("/export/runs")
|
||||
@@ -710,9 +716,7 @@ def export_runs(
|
||||
three fields rather than two blobs, and the defaults reach the same
|
||||
depth.
|
||||
"""
|
||||
runs = _selected(session, flow, status, group, ids, since, until)
|
||||
truncated = len(runs) > EXPORT_CAP
|
||||
runs = runs[:EXPORT_CAP]
|
||||
runs, total = _selected(session, flow, status, group, ids, since, until)
|
||||
inputs = [part for part in params.split(",") if part] or sorted(
|
||||
{path for run in runs for path, _ in _leaves(run.params)}
|
||||
)
|
||||
@@ -730,7 +734,7 @@ def export_runs(
|
||||
row.update((f"metric.{k}", _cell(_dig(run.result, k))) for k in scores)
|
||||
yield [row]
|
||||
|
||||
return _stream(format, "runs", columns, chunks(), truncated)
|
||||
return _stream(format, "runs", columns, chunks(), total)
|
||||
|
||||
|
||||
@router.get("/metrics/names", response_model=list[str])
|
||||
|
||||
@@ -17,6 +17,7 @@ import pkgutil
|
||||
import shutil
|
||||
import sys
|
||||
import time
|
||||
import warnings
|
||||
from collections.abc import Callable, Iterable, Iterator
|
||||
from contextlib import contextmanager, nullcontext
|
||||
from datetime import UTC, datetime
|
||||
@@ -1326,7 +1327,11 @@ def _export(
|
||||
try:
|
||||
if getattr(args, "list_names", False):
|
||||
return _list_names(client, args)
|
||||
rows = fetch(client)
|
||||
with warnings.catch_warnings(record=True) as caught:
|
||||
warnings.simplefilter("always")
|
||||
rows = fetch(client)
|
||||
for warning in caught:
|
||||
print(f"fluksio: warning: {warning.message}", file=sys.stderr)
|
||||
except ApiError as exc:
|
||||
if exc.status != 404:
|
||||
raise
|
||||
|
||||
@@ -12,6 +12,7 @@ import os
|
||||
import subprocess
|
||||
import time
|
||||
import uuid
|
||||
import warnings
|
||||
from collections.abc import Iterable
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
@@ -511,6 +512,8 @@ class Client:
|
||||
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.
|
||||
A selection past the engine's cap comes back without its oldest runs,
|
||||
and that is a warning, since the rows alone look complete.
|
||||
"""
|
||||
if flow:
|
||||
query["flow"] = flow
|
||||
@@ -523,6 +526,16 @@ class Client:
|
||||
response = self._request("GET", path, idempotent=True, params=query)
|
||||
if response.status_code >= 400:
|
||||
raise ApiError(response.status_code, _detail(response))
|
||||
cap = response.headers.get("X-Truncated")
|
||||
if cap:
|
||||
# An engine older than the count sends the cap alone.
|
||||
total = response.headers.get("X-Total-Count")
|
||||
left = f"{int(total) - int(cap)} older runs" if total else "older runs"
|
||||
warnings.warn(
|
||||
f"the export stops at {cap} runs, so {left} of this selection "
|
||||
"were left out; narrow it by flow, status, since or until",
|
||||
stacklevel=3,
|
||||
)
|
||||
return [json.loads(line) for line in response.text.splitlines() if line]
|
||||
|
||||
def cancel(self, run_id: str) -> Any:
|
||||
|
||||
@@ -1477,3 +1477,26 @@ def test_runs_page_by_when_they_were_created(client, superuser_token_headers, pa
|
||||
"page-2",
|
||||
"page-1",
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("paged")
|
||||
def test_a_truncated_export_says_how_many_runs_it_left_out(
|
||||
client, superuser_token_headers, monkeypatch
|
||||
):
|
||||
"""The cap is in one header, the selection's size in another, and the SDK
|
||||
turns the two into a warning naming what is missing."""
|
||||
from fluksio.sdk.client import Client
|
||||
|
||||
monkeypatch.setattr("fluksio.api.routes.runs.EXPORT_CAP", 2)
|
||||
answer = client.get(
|
||||
f"{settings.API_V1_STR}/runs/export/runs",
|
||||
params={"flow": "paged-study", "format": "jsonl"},
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert answer.headers["X-Truncated"] == "2"
|
||||
assert answer.headers["X-Total-Count"] == "3"
|
||||
assert [row["id"] for row in _lines(answer)] == ["page-2", "page-1"]
|
||||
|
||||
sdk = Client(http=client, token=superuser_token_headers["Authorization"][7:])
|
||||
with pytest.warns(UserWarning, match="1 older runs"):
|
||||
assert len(sdk.export_runs(flow="paged-study")) == 2
|
||||
|
||||
+3
-1
@@ -143,7 +143,9 @@ row's `created_at` as `before` for stable pagination.
|
||||
Exports also accept `ids` and `until`, with `format=csv` or `jsonl`.
|
||||
The metrics export is a long table; the runs export is one row per run.
|
||||
Use `params` and `metrics` to choose columns, including dotted paths into
|
||||
record outputs.
|
||||
record outputs. An export covers at most the newest 10 000 runs. When a
|
||||
selection matches more, the response carries `X-Truncated` with that cap and
|
||||
`X-Total-Count` with the number of runs the selection matched.
|
||||
|
||||
Import takes `{"runs": [...]}`, records completed results without executing
|
||||
code, and deduplicates by `external_id`. See
|
||||
|
||||
@@ -297,6 +297,8 @@ select columns, including dotted paths such as `final_metrics.train_loss`.
|
||||
|
||||
Both support `--flow`, repeated `--run ID`, `--group`, `--status`,
|
||||
`--since`, `--until` and `--local`. The run ID joins results to metrics.
|
||||
An export stops at the newest 10 000 runs, and when a selection matches more,
|
||||
the CLI and the Python client warn how many were left out.
|
||||
|
||||
Output defaults to CSV on stdout. `-o FILE` writes a file; `--format jsonl`
|
||||
writes JSON Lines. Parquet requires `pip install 'fluksio[parquet]'` and
|
||||
|
||||
Reference in New Issue
Block a user