Add a groups JSON API and group-run cancellation
The group pages were browser-only, so external schedulers had no way to
trigger a group or read its outcome. Adds /api/groups, /api/group-runs and
their run/list endpoints under Basic auth, following the same async contract
as POST /modules/{id}/run.
Cancellation needed more than an endpoint. run_group deliberately continues
past module failures, so killing the in-flight jrunner alone would just let
the loop march on to the next member. A group-level flag in engine/cancel.py
is now checked before each member, and engine.request_group_cancel sets that
flag *and* signals the running module -- it lives in runner.py rather than
cancel.py because finding the live run needs the DB, and cancel.py is
deliberately DB-free.
Group final status gains 'cancelled', ranked above 'error' since the error is
usually the terminated jrunner process. group_run.status already permitted the
value; no migration needed.
Also adds a "cancel group" button to the group run page (web) and .pill
styling for cancelled/dry_run, neither of which had a color rule.
Note: /runs/{id}/cancel keeps its existing meaning -- cancel one member and
let the group continue -- which is now a deliberate contrast to cancel group.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
parent
ae959aedad
commit
7de9d83589
@ -80,7 +80,9 @@ run_group(group_id)
|
|||||||
→ create group_run row (status=running)
|
→ create group_run row (status=running)
|
||||||
→ for each enabled group_member in run_order: call run_module(group_run_id=...)
|
→ for each enabled group_member in run_order: call run_module(group_run_id=...)
|
||||||
→ continues past individual module failures (all members run)
|
→ continues past individual module failures (all members run)
|
||||||
→ final status: dry_run if all dry_run, error if any errored, success otherwise
|
→ but a group cancel breaks the loop: remaining members never start
|
||||||
|
→ final status: cancelled if cancelled, else dry_run if all dry_run,
|
||||||
|
error if any errored, success otherwise
|
||||||
→ finish_group_run(status)
|
→ finish_group_run(status)
|
||||||
```
|
```
|
||||||
|
|
||||||
|
|||||||
@ -19,7 +19,7 @@ import os
|
|||||||
|
|
||||||
from .. import __version__, db, jrunner, repo
|
from .. import __version__, db, jrunner, repo
|
||||||
from ..web import mount_web
|
from ..web import mount_web
|
||||||
from .routes import connections, introspect, modules, runs, system
|
from .routes import connections, groups, introspect, modules, runs, system
|
||||||
|
|
||||||
_log = logging.getLogger("pipekit.web")
|
_log = logging.getLogger("pipekit.web")
|
||||||
|
|
||||||
@ -167,6 +167,7 @@ def create_app() -> FastAPI:
|
|||||||
app.include_router(system.router)
|
app.include_router(system.router)
|
||||||
app.include_router(connections.router, prefix="/api")
|
app.include_router(connections.router, prefix="/api")
|
||||||
app.include_router(introspect.router, prefix="/api")
|
app.include_router(introspect.router, prefix="/api")
|
||||||
|
app.include_router(groups.router, prefix="/api")
|
||||||
app.include_router(modules.router, prefix="/api")
|
app.include_router(modules.router, prefix="/api")
|
||||||
app.include_router(runs.router, prefix="/api")
|
app.include_router(runs.router, prefix="/api")
|
||||||
mount_web(app)
|
mount_web(app)
|
||||||
|
|||||||
84
pipekit/api/routes/groups.py
Normal file
84
pipekit/api/routes/groups.py
Normal file
@ -0,0 +1,84 @@
|
|||||||
|
"""Groups + group runs. The JSON counterpart to the web UI's group pages.
|
||||||
|
|
||||||
|
The web routes (web/app.py) already expose group runs, but behind session-cookie auth and
|
||||||
|
returning HTML/redirects, so they are only usable from a browser. External schedulers need
|
||||||
|
Basic auth and JSON -- hence these. Same async contract as POST /modules/{id}/run: the
|
||||||
|
group_run_id comes back immediately and the work continues in the background, so callers poll
|
||||||
|
GET /group-runs/{id} for the terminal status.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
|
||||||
|
|
||||||
|
from ... import engine, repo
|
||||||
|
from ..auth import require_auth
|
||||||
|
|
||||||
|
router = APIRouter(tags=["groups"], dependencies=[Depends(require_auth)])
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/groups")
|
||||||
|
def list_groups() -> list[dict]:
|
||||||
|
return repo.list_groups()
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/groups/{group_id}")
|
||||||
|
def get_group(group_id: int) -> dict:
|
||||||
|
g = repo.get_group(group_id)
|
||||||
|
if g is None:
|
||||||
|
raise HTTPException(404, f"group id={group_id} not found")
|
||||||
|
return {**g, "members": repo.list_group_members(group_id)}
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/groups/{group_id}/run")
|
||||||
|
def run_group(group_id: int, background: BackgroundTasks,
|
||||||
|
dry_run: bool = False) -> dict:
|
||||||
|
"""Kick off a group run. Returns group_run_id immediately."""
|
||||||
|
g = repo.get_group(group_id)
|
||||||
|
if g is None:
|
||||||
|
raise HTTPException(404, f"group id={group_id} not found")
|
||||||
|
group_run_id = repo.create_group_run(group_id, triggered_by="api")
|
||||||
|
background.add_task(_run_in_background, group_id, group_run_id, dry_run)
|
||||||
|
return {"group_run_id": group_run_id}
|
||||||
|
|
||||||
|
|
||||||
|
def _run_in_background(group_id: int, group_run_id: int, dry_run: bool) -> None:
|
||||||
|
try:
|
||||||
|
engine.run_group(group_id, group_run_id=group_run_id, dry_run=dry_run)
|
||||||
|
except Exception as e: # noqa: BLE001
|
||||||
|
# run_group already records per-module failures; this is the group row itself
|
||||||
|
# failing (or a LockBusy escaping), which must not leave it stuck at 'running'.
|
||||||
|
repo.finish_group_run(group_run_id, status="error")
|
||||||
|
raise e from None
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/group-runs/{group_run_id}")
|
||||||
|
def get_group_run(group_run_id: int) -> dict:
|
||||||
|
gr = repo.get_group_run(group_run_id)
|
||||||
|
if gr is None:
|
||||||
|
raise HTTPException(404, f"group_run id={group_run_id} not found")
|
||||||
|
return {**gr, "runs": repo.list_runs_for_group_run(group_run_id)}
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/group-runs/{group_run_id}/cancel")
|
||||||
|
def cancel_group_run(group_run_id: int) -> dict:
|
||||||
|
"""Stop an in-flight group run.
|
||||||
|
|
||||||
|
Kills the module currently running and prevents the remaining members from
|
||||||
|
starting. The group run lands with status ``cancelled``; the engine's own
|
||||||
|
finally-blocks release the module lock and write the run_log rows, so this
|
||||||
|
handler deliberately does not touch either.
|
||||||
|
"""
|
||||||
|
gr = repo.get_group_run(group_run_id)
|
||||||
|
if gr is None:
|
||||||
|
raise HTTPException(404, f"group_run id={group_run_id} not found")
|
||||||
|
if gr["status"] != "running":
|
||||||
|
raise HTTPException(409, f"group run {group_run_id} is {gr['status']}, not running")
|
||||||
|
return engine.request_group_cancel(group_run_id)
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/groups/{group_id}/group-runs")
|
||||||
|
def list_group_runs(group_id: int, limit: int = 20) -> list[dict]:
|
||||||
|
if repo.get_group(group_id) is None:
|
||||||
|
raise HTTPException(404, f"group id={group_id} not found")
|
||||||
|
return repo.list_group_runs(group_id, limit=limit)
|
||||||
@ -1,5 +1,6 @@
|
|||||||
from .cancel import request_cancel
|
from .cancel import request_cancel
|
||||||
from .runner import GroupRunOutcome, LockBusy, RunOutcome, run_group, run_module
|
from .runner import (GroupRunOutcome, LockBusy, RunOutcome, request_group_cancel,
|
||||||
|
run_group, run_module)
|
||||||
|
|
||||||
__all__ = ["GroupRunOutcome", "LockBusy", "RunOutcome", "request_cancel",
|
__all__ = ["GroupRunOutcome", "LockBusy", "RunOutcome", "request_cancel",
|
||||||
"run_group", "run_module"]
|
"request_group_cancel", "run_group", "run_module"]
|
||||||
|
|||||||
@ -17,6 +17,7 @@ import threading
|
|||||||
_lock = threading.Lock()
|
_lock = threading.Lock()
|
||||||
_active: dict[int, object] = {} # run_id -> subprocess.Popen (live jrunner proc)
|
_active: dict[int, object] = {} # run_id -> subprocess.Popen (live jrunner proc)
|
||||||
_cancelled: set[int] = set() # run_ids asked to stop
|
_cancelled: set[int] = set() # run_ids asked to stop
|
||||||
|
_cancelled_groups: set[int] = set() # group_run_ids asked to stop
|
||||||
|
|
||||||
|
|
||||||
class RunCancelled(Exception):
|
class RunCancelled(Exception):
|
||||||
@ -29,6 +30,27 @@ def arm(run_id: int) -> None:
|
|||||||
_cancelled.discard(run_id)
|
_cancelled.discard(run_id)
|
||||||
|
|
||||||
|
|
||||||
|
def arm_group(group_run_id: int) -> None:
|
||||||
|
"""Clear any stale group cancel flag before a group run starts."""
|
||||||
|
with _lock:
|
||||||
|
_cancelled_groups.discard(group_run_id)
|
||||||
|
|
||||||
|
|
||||||
|
def flag_group(group_run_id: int) -> None:
|
||||||
|
"""Mark a group run as cancelled so run_group stops launching members.
|
||||||
|
|
||||||
|
Only sets the flag; terminating the module that is already in flight is the
|
||||||
|
caller's job (it needs the DB to find the live run_id).
|
||||||
|
"""
|
||||||
|
with _lock:
|
||||||
|
_cancelled_groups.add(group_run_id)
|
||||||
|
|
||||||
|
|
||||||
|
def is_group_cancelled(group_run_id: int) -> bool:
|
||||||
|
with _lock:
|
||||||
|
return group_run_id in _cancelled_groups
|
||||||
|
|
||||||
|
|
||||||
def register(run_id: int, proc: object) -> None:
|
def register(run_id: int, proc: object) -> None:
|
||||||
"""Record the live subprocess for a run so cancel can reach it."""
|
"""Record the live subprocess for a run so cancel can reach it."""
|
||||||
with _lock:
|
with _lock:
|
||||||
|
|||||||
@ -38,7 +38,7 @@ class RunOutcome:
|
|||||||
@dataclass
|
@dataclass
|
||||||
class GroupRunOutcome:
|
class GroupRunOutcome:
|
||||||
group_run_id: int
|
group_run_id: int
|
||||||
status: str # success | error | dry_run
|
status: str # success | error | dry_run | cancelled
|
||||||
module_outcomes: list
|
module_outcomes: list
|
||||||
|
|
||||||
|
|
||||||
@ -208,10 +208,17 @@ def run_group(group_id: int, *, dry_run: bool = False,
|
|||||||
if group_run_id is None:
|
if group_run_id is None:
|
||||||
group_run_id = repo.create_group_run(group_id, triggered_by="manual")
|
group_run_id = repo.create_group_run(group_id, triggered_by="manual")
|
||||||
|
|
||||||
|
cancel.arm_group(group_run_id)
|
||||||
members = [m for m in repo.list_group_members(group_id) if m["module_enabled"]]
|
members = [m for m in repo.list_group_members(group_id) if m["module_enabled"]]
|
||||||
outcomes: list[RunOutcome] = []
|
outcomes: list[RunOutcome] = []
|
||||||
|
stopped = False
|
||||||
|
|
||||||
for member in members:
|
for member in members:
|
||||||
|
# Unlike a module failure (which the group runs past), a cancel means stop:
|
||||||
|
# don't start any remaining member.
|
||||||
|
if cancel.is_group_cancelled(group_run_id):
|
||||||
|
stopped = True
|
||||||
|
break
|
||||||
try:
|
try:
|
||||||
outcome = run_module(
|
outcome = run_module(
|
||||||
member["module_id"],
|
member["module_id"],
|
||||||
@ -224,7 +231,10 @@ def run_group(group_id: int, *, dry_run: bool = False,
|
|||||||
outcome = RunOutcome(run_id, "error", None, str(e), None, None)
|
outcome = RunOutcome(run_id, "error", None, str(e), None, None)
|
||||||
outcomes.append(outcome)
|
outcomes.append(outcome)
|
||||||
|
|
||||||
if not outcomes:
|
# 'cancelled' outranks 'error': the error is usually the killed jrunner proc.
|
||||||
|
if stopped or any(o.status == "cancelled" for o in outcomes):
|
||||||
|
final = "cancelled"
|
||||||
|
elif not outcomes:
|
||||||
final = "dry_run" if dry_run else "success"
|
final = "dry_run" if dry_run else "success"
|
||||||
elif all(o.status == "dry_run" for o in outcomes):
|
elif all(o.status == "dry_run" for o in outcomes):
|
||||||
final = "dry_run"
|
final = "dry_run"
|
||||||
@ -237,6 +247,23 @@ def run_group(group_id: int, *, dry_run: bool = False,
|
|||||||
return GroupRunOutcome(group_run_id, final, outcomes)
|
return GroupRunOutcome(group_run_id, final, outcomes)
|
||||||
|
|
||||||
|
|
||||||
|
def request_group_cancel(group_run_id: int) -> dict:
|
||||||
|
"""Stop a group run: flag it, and terminate whichever module is in flight.
|
||||||
|
|
||||||
|
Lives here rather than in cancel.py because finding the live run needs the DB.
|
||||||
|
The flag is what stops *remaining* members; the per-run cancel is what stops
|
||||||
|
the current one. Returns the runs that were signalled (may be empty if the
|
||||||
|
group is between members this instant -- the flag still lands).
|
||||||
|
"""
|
||||||
|
cancel.flag_group(group_run_id)
|
||||||
|
signalled = [
|
||||||
|
{"run_id": r["id"], "result": cancel.request_cancel(r["id"])}
|
||||||
|
for r in repo.list_runs_for_group_run(group_run_id)
|
||||||
|
if r["status"] == "running"
|
||||||
|
]
|
||||||
|
return {"group_run_id": group_run_id, "signalled": signalled}
|
||||||
|
|
||||||
|
|
||||||
def _run_hooks(module_id: int, *, fail_fast: bool, run_on_set: set[str]) -> str:
|
def _run_hooks(module_id: int, *, fail_fast: bool, run_on_set: set[str]) -> str:
|
||||||
"""Run hooks whose ``run_on`` is in run_on_set. Returns a text log."""
|
"""Run hooks whose ``run_on`` is in run_on_set. Returns a text log."""
|
||||||
hooks = [h for h in repo.list_hooks(module_id) if h["run_on"] in run_on_set]
|
hooks = [h for h in repo.list_hooks(module_id) if h["run_on"] in run_on_set]
|
||||||
|
|||||||
@ -1508,6 +1508,28 @@ def group_run_live_fragment(request: Request, group_run_id: int):
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@_router.post("/group-runs/{group_run_id}/cancel", response_class=HTMLResponse)
|
||||||
|
def group_run_cancel(request: Request, group_run_id: int):
|
||||||
|
"""Stop the whole group: kill the module in flight, skip the rest.
|
||||||
|
|
||||||
|
Distinct from /runs/{id}/cancel, which cancels one member and lets the group
|
||||||
|
carry on to the next.
|
||||||
|
"""
|
||||||
|
group_run = repo.get_group_run(group_run_id)
|
||||||
|
if group_run is None:
|
||||||
|
raise HTTPException(404, f"group run id={group_run_id} not found")
|
||||||
|
if group_run["status"] == "running":
|
||||||
|
engine.request_group_cancel(group_run_id)
|
||||||
|
# Re-read: the engine writes the terminal status from its own thread, so the
|
||||||
|
# fragment may still show 'running' for a tick until the poll catches up.
|
||||||
|
group_run = repo.get_group_run(group_run_id)
|
||||||
|
module_runs = repo.list_runs_for_group_run(group_run_id)
|
||||||
|
return _templates.TemplateResponse(
|
||||||
|
request, "_group_run_live.html",
|
||||||
|
_ctx(group_run=group_run, module_runs=module_runs),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def _save_group_schedules(form, group_id: int) -> None:
|
def _save_group_schedules(form, group_id: int) -> None:
|
||||||
"""Process sched_* parallel arrays from a group form POST."""
|
"""Process sched_* parallel arrays from a group form POST."""
|
||||||
from croniter import croniter as _croniter
|
from croniter import croniter as _croniter
|
||||||
|
|||||||
@ -147,6 +147,8 @@ table.grid tr:hover td { background: #1c2128; }
|
|||||||
.pill.running { color: var(--accent); }
|
.pill.running { color: var(--accent); }
|
||||||
.pill.disabled { color: var(--text-muted); }
|
.pill.disabled { color: var(--text-muted); }
|
||||||
.pill.warning { color: var(--warning); }
|
.pill.warning { color: var(--warning); }
|
||||||
|
.pill.cancelled { color: var(--warning); }
|
||||||
|
.pill.dry_run { color: var(--text-muted); }
|
||||||
|
|
||||||
/* Group membership tags */
|
/* Group membership tags */
|
||||||
.tag {
|
.tag {
|
||||||
|
|||||||
@ -9,6 +9,12 @@
|
|||||||
<header>
|
<header>
|
||||||
Module runs
|
Module runs
|
||||||
<span style="margin-left:auto">
|
<span style="margin-left:auto">
|
||||||
|
{% if group_run.status == 'running' %}
|
||||||
|
<button class="btn ghost" style="margin-right:0.5rem"
|
||||||
|
hx-post="/group-runs/{{ group_run.id }}/cancel"
|
||||||
|
hx-target="#group-run-live" hx-swap="outerHTML"
|
||||||
|
hx-confirm="Cancel this group run? The module in flight is stopped and the remaining modules are skipped.">cancel group</button>
|
||||||
|
{% endif %}
|
||||||
<span class="pill {{ group_run.status }}">{{ group_run.status }}</span>
|
<span class="pill {{ group_run.status }}">{{ group_run.status }}</span>
|
||||||
{% if group_run.duration_s is not none %} · {{ group_run.duration_s | duration }}{% endif %}
|
{% if group_run.duration_s is not none %} · {{ group_run.duration_s | duration }}{% endif %}
|
||||||
</span>
|
</span>
|
||||||
|
|||||||
@ -31,7 +31,17 @@
|
|||||||
hx-swap="outerHTML"
|
hx-swap="outerHTML"
|
||||||
{% endif %}>
|
{% endif %}>
|
||||||
<div class="panel">
|
<div class="panel">
|
||||||
<header>Module runs</header>
|
<header>
|
||||||
|
Module runs
|
||||||
|
{% if group_run.status == 'running' %}
|
||||||
|
<span style="margin-left:auto">
|
||||||
|
<button class="btn ghost"
|
||||||
|
hx-post="/group-runs/{{ group_run.id }}/cancel"
|
||||||
|
hx-target="#group-run-live" hx-swap="outerHTML"
|
||||||
|
hx-confirm="Cancel this group run? The module in flight is stopped and the remaining modules are skipped.">cancel group</button>
|
||||||
|
</span>
|
||||||
|
{% endif %}
|
||||||
|
</header>
|
||||||
<div class="body tight">
|
<div class="body tight">
|
||||||
{% if module_runs %}
|
{% if module_runs %}
|
||||||
<table class="grid">
|
<table class="grid">
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user