diff --git a/backend/fluksio/api/routes/runs.py b/backend/fluksio/api/routes/runs.py index e1c1993..56711e1 100644 --- a/backend/fluksio/api/routes/runs.py +++ b/backend/fluksio/api/routes/runs.py @@ -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]) diff --git a/backend/fluksio/sdk/cli.py b/backend/fluksio/sdk/cli.py index 7393d56..4c651a1 100644 --- a/backend/fluksio/sdk/cli.py +++ b/backend/fluksio/sdk/cli.py @@ -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 diff --git a/backend/fluksio/sdk/client.py b/backend/fluksio/sdk/client.py index 0ef0ded..fa71835 100644 --- a/backend/fluksio/sdk/client.py +++ b/backend/fluksio/sdk/client.py @@ -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: diff --git a/backend/tests/api/routes/test_runs.py b/backend/tests/api/routes/test_runs.py index 35bd238..e34f331 100644 --- a/backend/tests/api/routes/test_runs.py +++ b/backend/tests/api/routes/test_runs.py @@ -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 diff --git a/docs/code/api.md b/docs/code/api.md index e608e89..065205c 100644 --- a/docs/code/api.md +++ b/docs/code/api.md @@ -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 diff --git a/docs/code/cli.md b/docs/code/cli.md index 301d7cb..7810998 100644 --- a/docs/code/cli.md +++ b/docs/code/cli.md @@ -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