GRM-54: Integrate tea Gitea CLI for API interactions (#68)
This commit is contained in:
@@ -0,0 +1,323 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Thin Python wrapper around the ``tea`` Gitea CLI for CI/CD scripts.
|
||||
|
||||
This module provides a programmatic interface to the ``tea`` CLI tool,
|
||||
parsing JSON output for structured data. It is used by CI scripts to
|
||||
avoid hand-rolling HTTP requests and to leverage the official Gitea CLI
|
||||
for reliability.
|
||||
|
||||
The wrapper requires ``tea`` to be installed and configured (run
|
||||
``make setup`` which calls ``scripts/install_tools.py`` and
|
||||
``scripts/setup.py``).
|
||||
|
||||
Operations supported via tea:
|
||||
- Creating pull requests
|
||||
- Creating issues
|
||||
- Adding labels to issues/PRs
|
||||
- Creating labels
|
||||
- Merging pull requests
|
||||
- Creating releases
|
||||
- Posting reviews on PRs
|
||||
- Listing branches
|
||||
|
||||
Operations NOT supported via tea (still use GiteaClient):
|
||||
- Wiki page management
|
||||
- Commit status checks
|
||||
- Runner discovery
|
||||
- PR file/commit listing (tea has limited support)
|
||||
- Branch protection with detailed config (tea only has basic protect/unprotect)
|
||||
|
||||
Usage::
|
||||
|
||||
from scripts.gitea_cli import TeaCLI
|
||||
|
||||
tea = TeaCLI()
|
||||
tea.create_issue("owner/repo", title="Bug", body="Description", labels=["bug"])
|
||||
tea.add_label("owner/repo", 42, ["ready-to-merge"])
|
||||
tea.create_release("owner/repo", tag="v1.0.0", title="Release 1.0.0", body="Notes")
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import shutil
|
||||
import subprocess # nosec B404
|
||||
from typing import Any
|
||||
|
||||
|
||||
class TeaCLIError(Exception):
|
||||
"""Raised when a tea CLI command fails."""
|
||||
|
||||
|
||||
class TeaCLI:
|
||||
"""Wrapper around the ``tea`` Gitea CLI tool.
|
||||
|
||||
All methods parse JSON output from tea for structured access.
|
||||
Commands are run with ``--output json`` where structured data is expected.
|
||||
"""
|
||||
|
||||
def __init__(self, tea_bin: str | None = None, repo: str | None = None) -> None:
|
||||
"""Initialize the tea CLI wrapper.
|
||||
|
||||
Args:
|
||||
tea_bin: Path to the tea binary. If None, auto-detect via shutil.which.
|
||||
repo: Default repo in ``owner/name`` format for commands that need it.
|
||||
"""
|
||||
self._tea = tea_bin or shutil.which("tea") or "tea"
|
||||
self._repo = repo
|
||||
|
||||
def _run(self, args: list[str], json_output: bool = True) -> str:
|
||||
"""Run a tea command and return stdout.
|
||||
|
||||
Args:
|
||||
args: Command arguments (without the leading ``tea``).
|
||||
json_output: If True, append ``--output json`` to the command.
|
||||
|
||||
Returns:
|
||||
stdout as a string.
|
||||
|
||||
Raises:
|
||||
TeaCLIError: If the command fails.
|
||||
"""
|
||||
cmd = [self._tea, *args]
|
||||
if json_output:
|
||||
cmd.extend(["--output", "json"])
|
||||
result = subprocess.run( # nosec B603
|
||||
cmd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
raise TeaCLIError(
|
||||
f"tea command failed (rc={result.returncode}): {' '.join(args)}\nstderr: {result.stderr.strip()}"
|
||||
)
|
||||
return result.stdout.strip()
|
||||
|
||||
def _run_raw(self, args: list[str]) -> str:
|
||||
"""Run a tea command without JSON output and return stdout."""
|
||||
return self._run(args, json_output=False)
|
||||
|
||||
def _repo_arg(self, repo: str | None = None) -> list[str]:
|
||||
"""Build the --repo argument list."""
|
||||
target = repo or self._repo
|
||||
if target:
|
||||
return ["--repo", target]
|
||||
return []
|
||||
|
||||
# -- Issues --
|
||||
|
||||
def create_issue(
|
||||
self,
|
||||
repo: str,
|
||||
title: str,
|
||||
body: str = "",
|
||||
labels: list[str] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Create an issue and return the issue dict.
|
||||
|
||||
Args:
|
||||
repo: Repository in ``owner/name`` format.
|
||||
title: Issue title.
|
||||
body: Issue body (markdown).
|
||||
labels: List of label names to attach.
|
||||
|
||||
Returns:
|
||||
The created issue as a dict (parsed from tea JSON output).
|
||||
"""
|
||||
args = ["issues", "create", "--title", title, "--body", body, *self._repo_arg(repo)]
|
||||
output = self._run(args, json_output=False)
|
||||
# tea issues create doesn't output JSON; extract issue number from output
|
||||
# Format: "Created issue #42: <title>"
|
||||
issue_index = _extract_issue_number(output)
|
||||
return {"title": title, "body": body, "index": issue_index, "url": output.strip()}
|
||||
|
||||
# -- Labels --
|
||||
|
||||
def list_labels(self, repo: str) -> list[dict[str, Any]]:
|
||||
"""List all labels for a repository."""
|
||||
output = self._run(["labels", "list", *self._repo_arg(repo)])
|
||||
if not output:
|
||||
return []
|
||||
return json.loads(output)
|
||||
|
||||
def create_label(
|
||||
self,
|
||||
repo: str,
|
||||
name: str,
|
||||
color: str = "",
|
||||
description: str = "",
|
||||
) -> dict[str, Any]:
|
||||
"""Create a label. Returns the label dict.
|
||||
|
||||
Args:
|
||||
repo: Repository in ``owner/name`` format.
|
||||
name: Label name.
|
||||
color: Hex color (without #), e.g. ``2ecc71``.
|
||||
description: Label description.
|
||||
"""
|
||||
args = ["labels", "create", name, *self._repo_arg(repo)]
|
||||
if color:
|
||||
args.extend(["--color", f"#{color}"])
|
||||
if description:
|
||||
args.extend(["--description", description])
|
||||
output = self._run(args, json_output=False)
|
||||
return {"name": name, "color": color, "description": description, "output": output}
|
||||
|
||||
def add_label(self, repo: str, issue_index: int, labels: list[str]) -> None:
|
||||
"""Add labels to an issue or PR.
|
||||
|
||||
Args:
|
||||
repo: Repository in ``owner/name`` format.
|
||||
issue_index: Issue or PR number.
|
||||
labels: List of label names to add.
|
||||
"""
|
||||
for _label in labels:
|
||||
self._run_raw(["issues", "edit", "--add-labels", ",".join(labels), str(issue_index), *self._repo_arg(repo)])
|
||||
return # tea edit handles all labels at once
|
||||
# No labels to add — nothing to do
|
||||
|
||||
# -- Pull Requests --
|
||||
|
||||
def create_pr(
|
||||
self,
|
||||
repo: str,
|
||||
title: str,
|
||||
head: str,
|
||||
base: str,
|
||||
body: str = "",
|
||||
) -> dict[str, Any]:
|
||||
"""Create a pull request and return the PR dict.
|
||||
|
||||
Args:
|
||||
repo: Repository in ``owner/name`` format.
|
||||
title: PR title.
|
||||
head: Head branch name.
|
||||
base: Base branch name.
|
||||
body: PR description (markdown).
|
||||
"""
|
||||
args = [
|
||||
"pulls",
|
||||
"create",
|
||||
"--title",
|
||||
title,
|
||||
"--base",
|
||||
base,
|
||||
"--head",
|
||||
head,
|
||||
*self._repo_arg(repo),
|
||||
]
|
||||
if body:
|
||||
args.extend(["--body", body])
|
||||
output = self._run(args, json_output=False)
|
||||
pr_index = _extract_pr_number(output)
|
||||
return {"title": title, "index": pr_index, "url": output.strip()}
|
||||
|
||||
def merge_pr(self, repo: str, pr_index: int, style: str = "squash") -> None:
|
||||
"""Merge a pull request.
|
||||
|
||||
Args:
|
||||
repo: Repository in ``owner/name`` format.
|
||||
pr_index: PR number.
|
||||
style: Merge style: ``squash``, ``merge``, ``rebase``, ``rebase-edit``.
|
||||
"""
|
||||
self._run_raw(["pulls", "merge", "--style", style, str(pr_index), *self._repo_arg(repo)])
|
||||
|
||||
def review_pr(
|
||||
self,
|
||||
repo: str,
|
||||
pr_index: int,
|
||||
event: str = "COMMENT",
|
||||
body: str = "",
|
||||
) -> None:
|
||||
"""Post a review on a pull request.
|
||||
|
||||
Args:
|
||||
repo: Repository in ``owner/name`` format.
|
||||
pr_index: PR number.
|
||||
event: Review event: ``APPROVE``, ``REQUEST_CHANGES``, ``COMMENT``.
|
||||
body: Review body text.
|
||||
"""
|
||||
args = ["pulls", "review", str(pr_index), *self._repo_arg(repo)]
|
||||
if event == "APPROVE":
|
||||
args.append("--approve")
|
||||
elif event == "REQUEST_CHANGES":
|
||||
args.extend(["--reject"])
|
||||
if body:
|
||||
args.extend(["--comment", body])
|
||||
self._run_raw(args)
|
||||
|
||||
# -- Releases --
|
||||
|
||||
def create_release(
|
||||
self,
|
||||
repo: str,
|
||||
tag: str,
|
||||
title: str = "",
|
||||
body: str = "",
|
||||
target: str = "",
|
||||
draft: bool = False,
|
||||
prerelease: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
"""Create a release and return the release dict.
|
||||
|
||||
Args:
|
||||
repo: Repository in ``owner/name`` format.
|
||||
tag: Tag name (e.g. ``v1.0.0``).
|
||||
title: Release title.
|
||||
body: Release notes (markdown).
|
||||
target: Target branch/commit for the tag.
|
||||
draft: If True, create as draft.
|
||||
prerelease: If True, mark as prerelease.
|
||||
"""
|
||||
args = ["releases", "create", tag, *self._repo_arg(repo)]
|
||||
if title:
|
||||
args.extend(["--title", title])
|
||||
if body:
|
||||
args.extend(["--note", body])
|
||||
if target:
|
||||
args.extend(["--target", target])
|
||||
if draft:
|
||||
args.append("--draft")
|
||||
if prerelease:
|
||||
args.append("--prerelease")
|
||||
output = self._run(args, json_output=False)
|
||||
return {"tag": tag, "title": title, "url": output.strip()}
|
||||
|
||||
def list_releases(self, repo: str) -> list[dict[str, Any]]:
|
||||
"""List all releases for a repository."""
|
||||
output = self._run(["releases", "list", *self._repo_arg(repo)])
|
||||
if not output:
|
||||
return []
|
||||
return json.loads(output)
|
||||
|
||||
# -- Branches --
|
||||
|
||||
def list_branches(self, repo: str) -> list[dict[str, Any]]:
|
||||
"""List all branches for a repository."""
|
||||
output = self._run(["branches", "list", *self._repo_arg(repo)])
|
||||
if not output:
|
||||
return []
|
||||
return json.loads(output)
|
||||
|
||||
# -- Utility --
|
||||
|
||||
def whoami(self) -> str:
|
||||
"""Return the current authenticated user."""
|
||||
return self._run_raw(["whoami"])
|
||||
|
||||
|
||||
def _extract_issue_number(output: str) -> int:
|
||||
"""Extract the issue number from tea output like 'Created issue #42: ...'."""
|
||||
for part in output.split():
|
||||
if part.startswith("#"):
|
||||
try:
|
||||
return int(part[1:].rstrip(":"))
|
||||
except ValueError:
|
||||
continue
|
||||
return 0
|
||||
|
||||
|
||||
def _extract_pr_number(output: str) -> int:
|
||||
"""Extract the PR number from tea output like 'Created PR #42: ...'."""
|
||||
return _extract_issue_number(output)
|
||||
Reference in New Issue
Block a user