utils/todo_hooks/gh.py
Krzysztof kuhy Rudnicki aec96eefd3 Add todo_hooks closer, drift sweep, PR import
Python 3.14 stdlib package wiring git to the todo MCP server:
- closer (post-commit): runs the done_check of each Todo: trailer
  task and close_task's it only on exit 0. It FAILS OPEN: the commit
  already happened, so a down server, bad token or no gh prints one
  warning and exits 0 instead of failing the hook.
- sweep: nightly drift report (systemd units shipped, not enabled).
- import-pr: a merged PR's '## Todo:' items become tasks, deduped.
- A done_check runs only if its sha256 is in
  ~/.config/todo-hooks/trusted-checks; 'Check: manual' notes are
  never auto-closed.
- Hooks exposed in .pre-commit-hooks.yaml (post-commit, post-merge,
  post-rewrite).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-02 12:24:29 +01:00

70 lines
1.8 KiB
Python

"""Find the merged pull requests of a commit through the ``gh`` CLI."""
from __future__ import annotations
from dataclasses import dataclass
import json
import subprocess
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from collections.abc import Callable
GH_TIMEOUT = 20.0
class GhError(Exception):
"""gh missing, offline, failing or answering nonsense."""
@dataclass(frozen=True)
class PullRequest:
"""A pull request GitHub links to a commit."""
number: int
body: str
merged: bool
def run_gh(args: list[str]) -> str:
"""Run ``gh <args>``; stdout, or GhError on any failure."""
try:
done = subprocess.run(
["gh", *args],
check=False,
capture_output=True,
text=True,
timeout=GH_TIMEOUT,
stdin=subprocess.DEVNULL,
)
except FileNotFoundError as err:
msg = "gh not installed"
raise GhError(msg) from err
except subprocess.TimeoutExpired as err:
msg = "gh timed out"
raise GhError(msg) from err
if done.returncode != 0:
raise GhError(done.stderr.strip() or f"gh exit {done.returncode}")
return done.stdout
def merged_prs(
owner: str,
name: str,
sha: str,
run: Callable[[list[str]], str] | None = None,
) -> list[PullRequest]:
"""Pull requests GitHub links to commit `sha`."""
reply = (run or run_gh)(["api", f"repos/{owner}/{name}/commits/{sha}/pulls"])
try:
data = json.loads(reply)
return [
PullRequest(
int(pr["number"]), pr.get("body") or "", bool(pr.get("merged_at"))
)
for pr in data
]
except (ValueError, TypeError, KeyError) as err:
msg = "unparseable gh output"
raise GhError(msg) from err