324 lines
10 KiB
Python
324 lines
10 KiB
Python
#!/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)
|