jpmschweitzerandClaude Opus 4.6
19f5c93ad3
Ergonomic Python CLI wrapping parameterized SQLite queries for ticket management. Supports list, show, create, assign, sprint, deps, search, epics, children, count commands with JSON output. Replaces raw SQL access for Si and other agents. Also includes Sprint 1 "Run" creation and story assignments: - #77, #79 → client - #78, #236 → server - #81, #82, #83 → joint - 36 tickets updated to done (boilerplate work completed by merged PRs) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
354 lines
14 KiB
Python
Executable File
354 lines
14 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Ticket CLI — ergonomic interface to the project ticketing database.
|
|
|
|
Usage:
|
|
ticket list [--status S] [--priority P] [--epic N] [--sprint N] [--assigned A]
|
|
ticket show <id>
|
|
ticket done <id> [<id> ...]
|
|
ticket status <id> <new_status>
|
|
ticket assign <id> <agent>
|
|
ticket unassign <id>
|
|
ticket sprint [--active]
|
|
ticket sprint assign <id> <sprint_id>
|
|
ticket deps <id>
|
|
ticket search <keyword>
|
|
ticket create <type> <title> [--parent N] [--priority P] [--decision D]
|
|
ticket epics [--status S]
|
|
ticket children <id>
|
|
ticket count [--status S]
|
|
|
|
All output is JSON on stdout.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
CONFIG_PATH = SCRIPT_DIR / "config.json"
|
|
|
|
|
|
def load_config():
|
|
with open(CONFIG_PATH, "r") as f:
|
|
cfg = json.load(f)
|
|
db_path = (SCRIPT_DIR / cfg["sqlite_db"]).resolve()
|
|
cfg["sqlite_db_resolved"] = str(db_path)
|
|
return cfg
|
|
|
|
|
|
def get_connection(cfg):
|
|
conn = sqlite3.connect(cfg["sqlite_db_resolved"])
|
|
conn.execute("PRAGMA journal_mode=WAL;")
|
|
conn.execute("PRAGMA foreign_keys=ON;")
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
|
|
def query(conn, sql, params=()):
|
|
cursor = conn.execute(sql, params)
|
|
columns = [desc[0] for desc in cursor.description] if cursor.description else []
|
|
return [dict(zip(columns, row)) for row in cursor.fetchall()]
|
|
|
|
|
|
def execute(conn, sql, params=()):
|
|
cursor = conn.execute(sql, params)
|
|
conn.commit()
|
|
return cursor.rowcount
|
|
|
|
|
|
def out(data):
|
|
print(json.dumps(data, indent=2))
|
|
|
|
|
|
def parse_flags(args, known_flags):
|
|
"""Parse --flag value pairs from args, return (flags_dict, positional_args)."""
|
|
flags = {}
|
|
positional = []
|
|
i = 0
|
|
while i < len(args):
|
|
if args[i].startswith("--") and args[i][2:] in known_flags:
|
|
key = args[i][2:]
|
|
if i + 1 < len(args):
|
|
flags[key] = args[i + 1]
|
|
i += 2
|
|
else:
|
|
positional.append(args[i])
|
|
i += 1
|
|
else:
|
|
positional.append(args[i])
|
|
i += 1
|
|
return flags, positional
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Commands
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def cmd_list(conn, args):
|
|
flags, _ = parse_flags(args, ["status", "priority", "epic", "sprint", "assigned"])
|
|
conditions = []
|
|
params = []
|
|
if "status" in flags:
|
|
conditions.append("t.status = ?")
|
|
params.append(flags["status"])
|
|
if "priority" in flags:
|
|
conditions.append("t.priority = ?")
|
|
params.append(flags["priority"])
|
|
if "epic" in flags:
|
|
conditions.append("t.parent_id = ?")
|
|
params.append(int(flags["epic"]))
|
|
if "sprint" in flags:
|
|
conditions.append("t.sprint_id = ?")
|
|
params.append(int(flags["sprint"]))
|
|
if "assigned" in flags:
|
|
conditions.append("t.assigned_to = ?")
|
|
params.append(flags["assigned"])
|
|
where = " AND ".join(conditions) if conditions else "1=1"
|
|
sql = f"""SELECT t.id, t.type, t.title, t.status, t.priority, t.assigned_to,
|
|
t.parent_id, t.sprint_id
|
|
FROM tickets t WHERE {where}
|
|
ORDER BY
|
|
CASE t.priority WHEN 'critical' THEN 0 WHEN 'high' THEN 1
|
|
WHEN 'medium' THEN 2 ELSE 3 END, t.id"""
|
|
rows = query(conn, sql, tuple(params))
|
|
out({"ok": True, "count": len(rows), "rows": rows})
|
|
|
|
|
|
def cmd_show(conn, ticket_id):
|
|
rows = query(conn, """SELECT t.*, p.title as parent_title
|
|
FROM tickets t LEFT JOIN tickets p ON t.parent_id = p.id
|
|
WHERE t.id = ?""", (ticket_id,))
|
|
if not rows:
|
|
out({"ok": False, "error": f"Ticket #{ticket_id} not found"})
|
|
return
|
|
ticket = rows[0]
|
|
# Get children
|
|
children = query(conn, "SELECT id, title, status, priority FROM tickets WHERE parent_id = ? ORDER BY id", (ticket_id,))
|
|
# Get dependencies (what blocks this)
|
|
blockers = query(conn, """SELECT t.id, t.title, t.status FROM ticket_deps d
|
|
JOIN tickets t ON d.blocker_id = t.id
|
|
WHERE d.blocked_id = ?""", (ticket_id,))
|
|
# Get dependents (what this blocks)
|
|
blocks = query(conn, """SELECT t.id, t.title, t.status FROM ticket_deps d
|
|
JOIN tickets t ON d.blocked_id = t.id
|
|
WHERE d.blocker_id = ?""", (ticket_id,))
|
|
ticket["children"] = children
|
|
ticket["blocked_by"] = blockers
|
|
ticket["blocks"] = blocks
|
|
out({"ok": True, "ticket": ticket})
|
|
|
|
|
|
def cmd_done(conn, ids):
|
|
updated = 0
|
|
for tid in ids:
|
|
updated += execute(conn, "UPDATE tickets SET status='done', updated_at=datetime('now') WHERE id=?", (int(tid),))
|
|
out({"ok": True, "updated": updated, "ids": [int(i) for i in ids]})
|
|
|
|
|
|
def cmd_status(conn, ticket_id, new_status):
|
|
valid = ('backlog', 'ready', 'in_progress', 'review', 'done', 'cancelled')
|
|
if new_status not in valid:
|
|
out({"ok": False, "error": f"Invalid status '{new_status}'. Valid: {', '.join(valid)}"})
|
|
return
|
|
updated = execute(conn, "UPDATE tickets SET status=?, updated_at=datetime('now') WHERE id=?", (new_status, int(ticket_id)))
|
|
out({"ok": True, "updated": updated, "id": int(ticket_id), "status": new_status})
|
|
|
|
|
|
def cmd_assign(conn, ticket_id, agent):
|
|
updated = execute(conn, "UPDATE tickets SET assigned_to=?, updated_at=datetime('now') WHERE id=?", (agent, int(ticket_id)))
|
|
out({"ok": True, "updated": updated, "id": int(ticket_id), "assigned_to": agent})
|
|
|
|
|
|
def cmd_unassign(conn, ticket_id):
|
|
updated = execute(conn, "UPDATE tickets SET assigned_to=NULL, updated_at=datetime('now') WHERE id=?", (int(ticket_id),))
|
|
out({"ok": True, "updated": updated, "id": int(ticket_id), "assigned_to": None})
|
|
|
|
|
|
def cmd_sprint(conn, args):
|
|
flags, positional = parse_flags(args, ["active"])
|
|
if positional and positional[0] == "assign" and len(positional) >= 3:
|
|
ticket_id, sprint_id = int(positional[1]), int(positional[2])
|
|
updated = execute(conn, "UPDATE tickets SET sprint_id=?, updated_at=datetime('now') WHERE id=?", (sprint_id, ticket_id))
|
|
out({"ok": True, "updated": updated, "id": ticket_id, "sprint_id": sprint_id})
|
|
return
|
|
conditions = []
|
|
params = []
|
|
if "active" in flags:
|
|
conditions.append("s.status = 'active'")
|
|
where = " AND ".join(conditions) if conditions else "1=1"
|
|
sprints = query(conn, f"""SELECT s.*, COUNT(t.id) as ticket_count,
|
|
SUM(CASE WHEN t.status='done' THEN 1 ELSE 0 END) as done_count
|
|
FROM sprints s LEFT JOIN tickets t ON t.sprint_id = s.id
|
|
WHERE {where} GROUP BY s.id ORDER BY s.id DESC""", tuple(params))
|
|
out({"ok": True, "count": len(sprints), "sprints": sprints})
|
|
|
|
|
|
def cmd_deps(conn, ticket_id):
|
|
blockers = query(conn, """SELECT t.id, t.title, t.status, t.priority FROM ticket_deps d
|
|
JOIN tickets t ON d.blocker_id = t.id
|
|
WHERE d.blocked_id = ? ORDER BY t.id""", (int(ticket_id),))
|
|
blocks = query(conn, """SELECT t.id, t.title, t.status, t.priority FROM ticket_deps d
|
|
JOIN tickets t ON d.blocked_id = t.id
|
|
WHERE d.blocker_id = ? ORDER BY t.id""", (int(ticket_id),))
|
|
out({"ok": True, "id": int(ticket_id), "blocked_by": blockers, "blocks": blocks})
|
|
|
|
|
|
def cmd_search(conn, keyword):
|
|
rows = query(conn, """SELECT id, type, title, status, priority, assigned_to
|
|
FROM tickets WHERE title LIKE ? OR description LIKE ?
|
|
ORDER BY id""", (f"%{keyword}%", f"%{keyword}%"))
|
|
out({"ok": True, "count": len(rows), "rows": rows})
|
|
|
|
|
|
def cmd_create(conn, args):
|
|
flags, positional = parse_flags(args, ["parent", "priority", "decision"])
|
|
if len(positional) < 2:
|
|
out({"ok": False, "error": "Usage: ticket create <type> <title> [--parent N] [--priority P] [--decision D]"})
|
|
return
|
|
ticket_type = positional[0]
|
|
title = " ".join(positional[1:])
|
|
parent_id = int(flags["parent"]) if "parent" in flags else None
|
|
priority = flags.get("priority", "medium")
|
|
decision_ref = flags.get("decision")
|
|
conn.execute(
|
|
"INSERT INTO tickets (type, title, parent_id, priority, decision_ref) VALUES (?, ?, ?, ?, ?)",
|
|
(ticket_type, title, parent_id, priority, decision_ref))
|
|
conn.commit()
|
|
last_id = query(conn, "SELECT last_insert_rowid() as id")[0]["id"]
|
|
out({"ok": True, "id": last_id, "title": title})
|
|
|
|
|
|
def cmd_epics(conn, args):
|
|
flags, _ = parse_flags(args, ["status"])
|
|
conditions = ["t.type = 'epic'"]
|
|
params = []
|
|
if "status" in flags:
|
|
conditions.append("t.status = ?")
|
|
params.append(flags["status"])
|
|
where = " AND ".join(conditions)
|
|
rows = query(conn, f"""SELECT t.id, t.title, t.status, t.priority, t.assigned_to,
|
|
COUNT(c.id) as child_count,
|
|
SUM(CASE WHEN c.status='done' THEN 1 ELSE 0 END) as done_count
|
|
FROM tickets t LEFT JOIN tickets c ON c.parent_id = t.id
|
|
WHERE {where} GROUP BY t.id
|
|
ORDER BY CASE t.priority WHEN 'critical' THEN 0 WHEN 'high' THEN 1
|
|
WHEN 'medium' THEN 2 ELSE 3 END, t.id""", tuple(params))
|
|
out({"ok": True, "count": len(rows), "rows": rows})
|
|
|
|
|
|
def cmd_children(conn, ticket_id):
|
|
rows = query(conn, """SELECT id, type, title, status, priority, assigned_to
|
|
FROM tickets WHERE parent_id = ? ORDER BY id""", (int(ticket_id),))
|
|
out({"ok": True, "count": len(rows), "parent_id": int(ticket_id), "rows": rows})
|
|
|
|
|
|
def cmd_count(conn, args):
|
|
flags, _ = parse_flags(args, ["status"])
|
|
if "status" in flags:
|
|
rows = query(conn, "SELECT COUNT(*) as count FROM tickets WHERE status = ?", (flags["status"],))
|
|
else:
|
|
rows = query(conn, "SELECT status, COUNT(*) as count FROM tickets GROUP BY status ORDER BY count DESC")
|
|
out({"ok": True, "rows": rows})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
HELP = """ticket — project ticket CLI
|
|
|
|
Usage:
|
|
ticket list [--status S] [--priority P] [--epic N] [--sprint N] [--assigned A]
|
|
ticket show <id> Full ticket detail with deps and children
|
|
ticket done <id> [<id> ...] Mark tickets as done
|
|
ticket status <id> <new_status> Change ticket status
|
|
ticket assign <id> <agent> Assign ticket to agent/branch
|
|
ticket unassign <id> Remove assignment
|
|
ticket sprint [--active] List sprints
|
|
ticket sprint assign <id> <sprint> Assign ticket to sprint
|
|
ticket deps <id> Show ticket dependencies
|
|
ticket search <keyword> Search tickets by title/description
|
|
ticket create <type> <title> [--parent N] [--priority P] [--decision D]
|
|
ticket epics [--status S] List epics with child counts
|
|
ticket children <id> List children of a ticket
|
|
ticket count [--status S] Count tickets by status"""
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) < 2 or sys.argv[1] in ("--help", "-h", "help"):
|
|
print(HELP)
|
|
sys.exit(0)
|
|
|
|
try:
|
|
cfg = load_config()
|
|
except (FileNotFoundError, json.JSONDecodeError) as exc:
|
|
out({"ok": False, "error": f"Config error: {exc}"})
|
|
sys.exit(1)
|
|
|
|
conn = get_connection(cfg)
|
|
cmd = sys.argv[1]
|
|
args = sys.argv[2:]
|
|
|
|
try:
|
|
if cmd == "list":
|
|
cmd_list(conn, args)
|
|
elif cmd == "show":
|
|
if not args:
|
|
out({"ok": False, "error": "Usage: ticket show <id>"})
|
|
else:
|
|
cmd_show(conn, int(args[0]))
|
|
elif cmd == "done":
|
|
if not args:
|
|
out({"ok": False, "error": "Usage: ticket done <id> [<id> ...]"})
|
|
else:
|
|
cmd_done(conn, args)
|
|
elif cmd == "status":
|
|
if len(args) < 2:
|
|
out({"ok": False, "error": "Usage: ticket status <id> <new_status>"})
|
|
else:
|
|
cmd_status(conn, args[0], args[1])
|
|
elif cmd == "assign":
|
|
if len(args) < 2:
|
|
out({"ok": False, "error": "Usage: ticket assign <id> <agent>"})
|
|
else:
|
|
cmd_assign(conn, args[0], args[1])
|
|
elif cmd == "unassign":
|
|
if not args:
|
|
out({"ok": False, "error": "Usage: ticket unassign <id>"})
|
|
else:
|
|
cmd_unassign(conn, args[0])
|
|
elif cmd == "sprint":
|
|
cmd_sprint(conn, args)
|
|
elif cmd == "deps":
|
|
if not args:
|
|
out({"ok": False, "error": "Usage: ticket deps <id>"})
|
|
else:
|
|
cmd_deps(conn, args[0])
|
|
elif cmd == "search":
|
|
if not args:
|
|
out({"ok": False, "error": "Usage: ticket search <keyword>"})
|
|
else:
|
|
cmd_search(conn, " ".join(args))
|
|
elif cmd == "create":
|
|
cmd_create(conn, args)
|
|
elif cmd == "epics":
|
|
cmd_epics(conn, args)
|
|
elif cmd == "children":
|
|
if not args:
|
|
out({"ok": False, "error": "Usage: ticket children <id>"})
|
|
else:
|
|
cmd_children(conn, args[0])
|
|
elif cmd == "count":
|
|
cmd_count(conn, args)
|
|
else:
|
|
out({"ok": False, "error": f"Unknown command: {cmd}. Use --help for usage."})
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|