pt/pipekit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add pipekit run-group for triggering a group from outside

Browse Source 
engine.run_group already sequences a group's enabled members by run_order,
continues past individual failures, and aggregates status -- but nothing
outside the process could reach it. The scheduler called it on cron, and
the web UI exposed it as a form POST that returns 303 + a background task.
Neither is usable from an external orchestrator: driving the web route
means scraping HTML to poll for completion.

This wraps it as a synchronous CLI command, mirroring `pipekit run` for
modules, so callers get an exit code and a summary on stdout.

LockBusy maps to exit 75 (EX_TEMPFAIL) rather than 1, so a caller can
distinguish "another run holds the lock, retry later" from a genuine
failure. This intentionally diverges from `pipekit run`, which returns 1
for both.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Paul Trowbridge 2026-08-19 17:32:24 -04:00
parent 4ae0b99108
commit e4c17bfc98
1 changed files with 27 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
pipekit/cli.py
View File

@ -139,6 +139,26 @@ def cmd_run(args) -> int:
return 0 if outcome.status == "success" else 1
def cmd_run_group(args) -> int:
group = repo.get_group_by_name(args.group)
if group is None:
print(f"error: group {args.group!r} not found")
return 1
try:
outcome = engine.run_group(group["id"], dry_run=args.dry_run)
except engine.LockBusy as e:
print(f"busy: {e}")
return 75 # EX_TEMPFAIL — retryable, distinct from a real failure
tag = "DRY RUN — no jrunner calls made" if args.dry_run else ""
print(f"group_run_id={outcome.group_run_id} status={outcome.status} "
f"modules={len(outcome.module_outcomes)} {tag}".rstrip())
for o in outcome.module_outcomes:
line = f" run_id={o.run_id} status={o.status} rows={o.row_count}"
print(f"{line} error={o.error}" if o.error else line)
return 0 if outcome.status in ("success", "dry_run") else 1
def cmd_serve(args) -> int:
import uvicorn
from .api import create_app
@ -351,6 +371,13 @@ def main(argv: list[str] | None = None) -> int:
help="build SQL but do not invoke jrunner")
p_run.set_defaults(func=cmd_run)
p_rg = sub.add_parser("run-group",
help="run all enabled modules in a group (synchronous)")
p_rg.add_argument("group", help="group name")
p_rg.add_argument("--dry-run", action="store_true",
help="build SQL but do not invoke jrunner")
p_rg.set_defaults(func=cmd_run_group)
p_exp = sub.add_parser(
"export", help="dump config (drivers/connections/modules/groups) to text files")
p_exp.add_argument("--dir", help="target dir (default <repo>/config)")