GET /runs/{id}/nodes/{node}/outputs serves what a node returned
The stage cache's stored outputs, as kept (up to 256 kB of JSON). A node that kept none answers 404 saying why. Listed in the API docs.
This commit is contained in:
@@ -27,7 +27,7 @@ from fluksio.core.config import settings
|
||||
from fluksio.core.db import writing
|
||||
from fluksio.flow.events import event_bus
|
||||
from fluksio.flow.messages import requalify
|
||||
from fluksio.flow.runs import RunRejected, RunService, new_run_id
|
||||
from fluksio.flow.runs import OUTPUT_CAP, RunRejected, RunService, new_run_id
|
||||
from fluksio.flow.store import FlowNotFound
|
||||
from fluksio.models import Message, Run, RunArtifact, RunMetric, RunNode
|
||||
|
||||
@@ -800,6 +800,28 @@ def read_run(run_id: str, session: SessionDep) -> Any:
|
||||
return detail
|
||||
|
||||
|
||||
@router.get("/{run_id}/nodes/{node}/outputs", response_model=dict[str, Any] | None)
|
||||
def read_node_outputs(run_id: str, node: str, session: SessionDep) -> Any:
|
||||
"""What one node returned, by message, as the stage cache stored it.
|
||||
|
||||
Kept only for a node that succeeded and may be reused, and only up to
|
||||
256 kB of JSON. Null when the node published nothing.
|
||||
"""
|
||||
row = session.get(RunNode, {"run_id": run_id, "node": node})
|
||||
if row is None:
|
||||
raise HTTPException(status_code=404, detail=f"No node '{node}' in this run")
|
||||
if row.outputs is None:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail=(
|
||||
f"Node '{node}' kept no outputs. It failed, may not be cached, "
|
||||
f"or returned more than {OUTPUT_CAP // 1000} kB or a value JSON "
|
||||
"cannot carry."
|
||||
),
|
||||
)
|
||||
return json.loads(row.outputs)
|
||||
|
||||
|
||||
@router.post("/{run_id}/cancel", response_model=RunRow)
|
||||
async def cancel_run(run_id: str, request: Request, session: SessionDep) -> Any:
|
||||
"""Stop a run. One already past its last node is left as it finished."""
|
||||
|
||||
@@ -1115,6 +1115,37 @@ def test_a_curve_whose_run_is_gone_is_empty_rather_than_an_error(
|
||||
assert node["curve_lost"] is True
|
||||
|
||||
|
||||
def test_a_node_answers_with_what_it_returned(client, superuser_token_headers):
|
||||
"""What the stage cache stored, and a 404 that says why for a node without."""
|
||||
run_id = new_run_id()
|
||||
with Session(db_engine) as session:
|
||||
session.add(
|
||||
Run(id=run_id, flow="study", status="ok", created_at=datetime.now(UTC))
|
||||
)
|
||||
session.add(RunNode(run_id=run_id, node="fit", outputs='{"study.loss":0.5}'))
|
||||
session.add(RunNode(run_id=run_id, node="plot", status="error"))
|
||||
session.commit()
|
||||
url = f"{settings.API_V1_STR}/runs/{run_id}/nodes"
|
||||
try:
|
||||
kept = client.get(f"{url}/fit/outputs", headers=superuser_token_headers)
|
||||
assert kept.json() == {"study.loss": 0.5}
|
||||
|
||||
none = client.get(f"{url}/plot/outputs", headers=superuser_token_headers)
|
||||
assert none.status_code == 404
|
||||
assert "kept no outputs" in none.json()["detail"]
|
||||
|
||||
absent = client.get(f"{url}/nope/outputs", headers=superuser_token_headers)
|
||||
assert absent.status_code == 404
|
||||
finally:
|
||||
with Session(db_engine) as session:
|
||||
for row in session.exec(
|
||||
select(RunNode).where(col(RunNode.run_id) == run_id)
|
||||
).all():
|
||||
session.delete(row)
|
||||
session.delete(session.get(Run, run_id))
|
||||
session.commit()
|
||||
|
||||
|
||||
def _metric(run_id: str, name: str, step: int, value: float, ts: float) -> RunMetric:
|
||||
return RunMetric(run_id=run_id, name=name, step=step, value=value, ts=ts)
|
||||
|
||||
|
||||
@@ -128,6 +128,7 @@ values. You can publish only to declared messages, just as dashboard controls do
|
||||
| `GET` | `/runs` | query history |
|
||||
| `GET` | `/runs/overview` | summarize runs per flow |
|
||||
| `GET` | `/runs/{id}` | read parameters, results, nodes, and artifacts |
|
||||
| `GET` | `/runs/{id}/nodes/{node}/outputs` | read what one node returned |
|
||||
| `POST` | `/runs/{id}/cancel`, `/runs/{id}/retry` | cancel, or create a retry |
|
||||
| `DELETE` | `/runs/{id}` | delete a finished run and its records |
|
||||
| `DELETE` | `/runs?flow={name}` | delete a flow's runs; refused while any are active |
|
||||
@@ -144,6 +145,10 @@ History filters include `flow`, `status`, `group`, `digest`, `since`,
|
||||
`before`, `limit`, and `offset`. Rows arrive newest first; use the last
|
||||
row's `created_at` as `before` for stable pagination.
|
||||
|
||||
A node's outputs come from the stage cache, so only a node that succeeded
|
||||
and may be cached has them, up to 256 kB of JSON. Any other node answers
|
||||
`404`.
|
||||
|
||||
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
|
||||
|
||||
File diff suppressed because one or more lines are too long
@@ -2162,6 +2162,15 @@ export type RunsDeleteRunData = {
|
||||
|
||||
export type RunsDeleteRunResponse = (void);
|
||||
|
||||
export type RunsReadNodeOutputsData = {
|
||||
node: string;
|
||||
runId: string;
|
||||
};
|
||||
|
||||
export type RunsReadNodeOutputsResponse = (({
|
||||
[key: string]: unknown;
|
||||
} | null));
|
||||
|
||||
export type RunsCancelRunData = {
|
||||
runId: string;
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user