338 lines
11 KiB
Python
338 lines
11 KiB
Python
"""Runs over the API: submit one, watch it, read what it made.
|
|
|
|
Submitting returns immediately with a queued run — a training run is measured
|
|
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.
|
|
"""
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from fastapi.concurrency import run_in_threadpool
|
|
from pydantic import BaseModel, Field
|
|
from sqlalchemy import func
|
|
from sqlalchemy import select as sa_select
|
|
from sqlmodel import col, select
|
|
|
|
from fluksio.api.deps import CurrentUser, SessionDep, get_current_user
|
|
from fluksio.flow.runs import RunRejected, RunService, new_run_id
|
|
from fluksio.flow.store import FlowNotFound
|
|
from fluksio.models import Run, RunArtifact, RunMetric, RunNode
|
|
|
|
router = APIRouter(
|
|
prefix="/runs", tags=["runs"], dependencies=[Depends(get_current_user)]
|
|
)
|
|
|
|
#: A sweep bigger than this is almost always a mistake in a loop.
|
|
MAX_SWEEP = 1000
|
|
|
|
|
|
class RunCreate(BaseModel):
|
|
params: dict[str, Any] = Field(default_factory=dict)
|
|
seed: int | None = None
|
|
#: Run the unpublished draft instead of what is published.
|
|
draft: bool = False
|
|
#: Execute every node, whatever an earlier run already worked out.
|
|
no_cache: bool = False
|
|
|
|
|
|
class SweepEntry(BaseModel):
|
|
params: dict[str, Any] = Field(default_factory=dict)
|
|
seed: int | None = None
|
|
|
|
|
|
class SweepCreate(BaseModel):
|
|
runs: list[SweepEntry] = Field(default_factory=list)
|
|
draft: bool = False
|
|
no_cache: bool = False
|
|
|
|
|
|
class RunNodeRow(BaseModel):
|
|
node: str
|
|
status: str
|
|
attempt: int
|
|
duration_ms: float
|
|
worker: str
|
|
error: str
|
|
logs: str
|
|
#: What this node's result was looked up by. Empty when it may not be
|
|
#: reused; `status` is "cached" when it was.
|
|
cache_key: str = ""
|
|
|
|
|
|
class ArtifactRow(BaseModel):
|
|
name: str
|
|
node: str
|
|
digest: str
|
|
size: int
|
|
media_type: str
|
|
|
|
|
|
class RunRow(BaseModel):
|
|
"""A run without its result, which is the part that can be large."""
|
|
|
|
id: str
|
|
flow: str
|
|
status: str
|
|
status_reason: str
|
|
cause: str
|
|
params: dict[str, Any]
|
|
params_digest: str
|
|
#: The user repository's commit, for a flow declared in code with the
|
|
#: decorators. Empty for one drawn on the canvas, where `commit` is the
|
|
#: whole answer to what produced the number.
|
|
origin_commit: str = ""
|
|
seed: int | None
|
|
group_id: str | None
|
|
labels: list[str]
|
|
created_at: Any
|
|
started_at: Any = None
|
|
finished_at: Any = None
|
|
duration_ms: float
|
|
actor: str
|
|
|
|
|
|
class RunDetail(RunRow):
|
|
result: dict[str, Any] = Field(default_factory=dict)
|
|
commit: str = ""
|
|
flow_version: int = 1
|
|
nodes: list[RunNodeRow] = Field(default_factory=list)
|
|
artifacts: list[ArtifactRow] = Field(default_factory=list)
|
|
|
|
|
|
class FlowRunsRow(BaseModel):
|
|
"""How much a flow has been run, for the screen's list of flows."""
|
|
|
|
flow: str
|
|
runs: int
|
|
running: int
|
|
queued: int
|
|
last_created_at: Any = None
|
|
|
|
|
|
class MetricPoint(BaseModel):
|
|
step: int
|
|
ts: float
|
|
value: float
|
|
name: str = ""
|
|
|
|
|
|
class MetricSeries(BaseModel):
|
|
"""The shape a chart widget already draws, so comparing runs is a binding."""
|
|
|
|
label: str
|
|
points: list[list[float]] = Field(default_factory=list)
|
|
|
|
|
|
class SeriesAnswer(BaseModel):
|
|
metric: str
|
|
lines: list[MetricSeries] = Field(default_factory=list)
|
|
|
|
|
|
def _service(request: Request) -> RunService:
|
|
service: RunService | None = getattr(request.app.state, "run_service", None)
|
|
if service is None:
|
|
raise HTTPException(status_code=503, detail="Runs are not available")
|
|
return service
|
|
|
|
|
|
@router.post("/flows/{name}", response_model=RunRow, status_code=202)
|
|
async def create_run(
|
|
name: str, body: RunCreate, request: Request, user: CurrentUser
|
|
) -> Any:
|
|
"""Queue one run of a flow."""
|
|
service = _service(request)
|
|
try:
|
|
return await run_in_threadpool(
|
|
service.submit,
|
|
name,
|
|
params=body.params,
|
|
seed=body.seed,
|
|
cause="api",
|
|
actor=user.email,
|
|
draft=body.draft,
|
|
no_cache=body.no_cache,
|
|
)
|
|
except FlowNotFound as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
|
except RunRejected as exc:
|
|
raise HTTPException(status_code=422, detail=str(exc)) from exc
|
|
|
|
|
|
@router.post("/flows/{name}/sweep", response_model=list[RunRow], status_code=202)
|
|
async def create_sweep(
|
|
name: str, body: SweepCreate, request: Request, user: CurrentUser
|
|
) -> Any:
|
|
"""Queue many runs of one flow under a shared group.
|
|
|
|
An ensemble is this with the same parameters and different seeds; a grid
|
|
search is this with the parameters spread out. Either way the caller
|
|
builds the list — the engine does not own a sweep grammar.
|
|
"""
|
|
if not body.runs:
|
|
raise HTTPException(status_code=422, detail="A sweep needs at least one run")
|
|
if len(body.runs) > MAX_SWEEP:
|
|
raise HTTPException(
|
|
status_code=422, detail=f"A sweep is capped at {MAX_SWEEP} runs"
|
|
)
|
|
service = _service(request)
|
|
group = new_run_id()
|
|
|
|
def submit_all() -> list[Run]:
|
|
return [
|
|
service.submit(
|
|
name,
|
|
params=entry.params,
|
|
seed=entry.seed,
|
|
group_id=group,
|
|
cause="sweep",
|
|
actor=user.email,
|
|
draft=body.draft,
|
|
no_cache=body.no_cache,
|
|
)
|
|
for entry in body.runs
|
|
]
|
|
|
|
try:
|
|
return await run_in_threadpool(submit_all)
|
|
except FlowNotFound as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
|
except RunRejected as exc:
|
|
raise HTTPException(status_code=422, detail=str(exc)) from exc
|
|
|
|
|
|
@router.get("", response_model=list[RunRow])
|
|
def read_runs(
|
|
session: SessionDep,
|
|
flow: str | None = None,
|
|
status: str | None = None,
|
|
group: str | None = None,
|
|
digest: str | None = None,
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
) -> Any:
|
|
"""Runs, newest first. The queryable table an experiment log needs."""
|
|
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)
|
|
if digest:
|
|
statement = statement.where(col(Run.params_digest) == digest)
|
|
statement = statement.offset(max(0, offset)).limit(min(limit, 500))
|
|
return list(session.exec(statement))
|
|
|
|
|
|
@router.get("/overview", response_model=list[FlowRunsRow])
|
|
def read_overview(session: SessionDep) -> Any:
|
|
"""One row per flow that has ever run, busiest-recent first.
|
|
|
|
The list caps at 500 newest runs, so counting flows on the client goes
|
|
wrong the moment a history outgrows one page. The database counts instead.
|
|
"""
|
|
statement = sa_select(
|
|
col(Run.flow),
|
|
col(Run.status),
|
|
func.count(col(Run.id)),
|
|
func.max(col(Run.created_at)),
|
|
).group_by(col(Run.flow), col(Run.status))
|
|
|
|
rows: dict[str, FlowRunsRow] = {}
|
|
for flow, status, count, latest in session.execute(statement):
|
|
row = rows.setdefault(flow, FlowRunsRow(flow=flow, runs=0, running=0, queued=0))
|
|
row.runs += count
|
|
if status == "running":
|
|
row.running += count
|
|
elif status == "queued":
|
|
row.queued += count
|
|
if row.last_created_at is None or latest > row.last_created_at:
|
|
row.last_created_at = latest
|
|
return sorted(rows.values(), key=lambda row: row.last_created_at, reverse=True)
|
|
|
|
|
|
@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."""
|
|
run = session.get(Run, run_id)
|
|
if run is None:
|
|
raise HTTPException(status_code=404, detail="No such run")
|
|
nodes = session.exec(select(RunNode).where(col(RunNode.run_id) == run_id)).all()
|
|
artifacts = session.exec(
|
|
select(RunArtifact).where(col(RunArtifact.run_id) == run_id)
|
|
).all()
|
|
detail = RunDetail.model_validate(run, from_attributes=True)
|
|
detail.nodes = [RunNodeRow.model_validate(n, from_attributes=True) for n in nodes]
|
|
detail.artifacts = [
|
|
ArtifactRow.model_validate(a, from_attributes=True) for a in artifacts
|
|
]
|
|
return detail
|
|
|
|
|
|
@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."""
|
|
run = session.get(Run, run_id)
|
|
if run is None:
|
|
raise HTTPException(status_code=404, detail="No such run")
|
|
service = _service(request)
|
|
await run_in_threadpool(service.cancel, run_id)
|
|
session.refresh(run)
|
|
return run
|
|
|
|
|
|
@router.get("/{run_id}/metrics", response_model=list[MetricPoint])
|
|
def read_metrics(
|
|
run_id: str, session: SessionDep, name: str = "", stride: int = 1
|
|
) -> Any:
|
|
"""One metric's series, in step order — or every one of them, unnamed.
|
|
|
|
``stride`` thins a long curve down: 3000 steps drawn on a 400-pixel chart
|
|
is 3000 points nobody can see.
|
|
"""
|
|
statement = select(RunMetric).where(col(RunMetric.run_id) == run_id)
|
|
if name:
|
|
statement = statement.where(col(RunMetric.name) == name)
|
|
statement = statement.order_by(col(RunMetric.name), col(RunMetric.step))
|
|
rows = list(session.exec(statement))
|
|
if stride > 1:
|
|
rows = rows[:: max(1, stride)]
|
|
return rows
|
|
|
|
|
|
@router.get("/series/compare", response_model=SeriesAnswer)
|
|
def compare_metric(session: SessionDep, ids: str, metric: str) -> Any:
|
|
"""One metric across several runs, as the chart widget's series shape.
|
|
|
|
This is the comparison view: it answers in the same shape a flow answers a
|
|
chart's query with, so putting three training curves beside each other is
|
|
a widget binding rather than a screen of its own.
|
|
"""
|
|
run_ids = [part for part in ids.split(",") if part]
|
|
if not run_ids:
|
|
raise HTTPException(status_code=422, detail="Name at least one run")
|
|
runs = {
|
|
run.id: run
|
|
for run in session.exec(select(Run).where(col(Run.id).in_(run_ids))).all()
|
|
}
|
|
lines: list[MetricSeries] = []
|
|
for run_id in run_ids:
|
|
run = runs.get(run_id)
|
|
if run is None:
|
|
continue
|
|
rows = session.exec(
|
|
select(RunMetric)
|
|
.where(col(RunMetric.run_id) == run_id, col(RunMetric.name) == metric)
|
|
.order_by(col(RunMetric.step))
|
|
).all()
|
|
label = run_id
|
|
if run.seed is not None:
|
|
label = f"{run_id} (seed {run.seed})"
|
|
lines.append(
|
|
MetricSeries(
|
|
label=label, points=[[float(row.step), row.value] for row in rows]
|
|
)
|
|
)
|
|
return SeriesAnswer(metric=metric, lines=lines)
|