Public Access
441 lines
14 KiB
Python
441 lines
14 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 devx.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 logging
|
|
import shutil
|
|
import subprocess # nosec B404
|
|
from typing import Any
|
|
|
|
import click
|
|
from tenacity import (
|
|
before_sleep_log,
|
|
retry,
|
|
retry_if_exception_type,
|
|
stop_after_attempt,
|
|
wait_exponential,
|
|
)
|
|
|
|
from devx.config import GITEA_API_URL, MAX_RETRIES, RETRY_BACKOFF_BASE, RETRY_STATUS_CODES
|
|
from devx.i18n import _
|
|
from devx.tokens import get_ci_token
|
|
|
|
logger = logging.getLogger("gitea_cli")
|
|
|
|
|
|
class TeaCLIError(Exception):
|
|
"""Raised when a tea CLI command fails."""
|
|
|
|
|
|
class _TransientTeaError(TeaCLIError):
|
|
"""Tea CLI error caused by a transient HTTP status (502/503/504/429)."""
|
|
|
|
|
|
def configure_tea_login(login_name: str = "devx") -> None:
|
|
"""Configure tea CLI login from CI_GITEA_API_TOKEN and DEVX_GITEA_API_URL.
|
|
|
|
Idempotent: if a login with the same name already exists, it is not re-added.
|
|
Skips silently if tea is not installed or no token is set.
|
|
|
|
Raises ``TeaCLIError`` if the login add or default command fails. This is
|
|
critical because subsequent tea commands (e.g. ``releases create``) will
|
|
fail with a cryptic "no available login" error if the login was not
|
|
configured successfully.
|
|
|
|
Used by CI scripts (publish, notify_failure) that need tea login but
|
|
run in containerized environments where ``make setup`` was not called.
|
|
"""
|
|
tea_bin = shutil.which("tea")
|
|
if tea_bin is None:
|
|
click.echo(_("tea not installed — skipping login configuration."))
|
|
return
|
|
|
|
try:
|
|
token = get_ci_token()
|
|
except click.ClickException:
|
|
click.echo(_("CI_GITEA_TOKEN not set — skipping login configuration."))
|
|
return
|
|
|
|
gitea_url = GITEA_API_URL.replace("/api/v1", "")
|
|
|
|
result = subprocess.run( # nosec B603
|
|
[tea_bin, "login", "list", "--output", "simple"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
if result.returncode == 0 and login_name in result.stdout:
|
|
click.echo(_("tea login '{name}' already configured.", name=login_name))
|
|
return
|
|
|
|
click.echo(_("Configuring tea login '{name}' for {url}...", name=login_name, url=gitea_url))
|
|
add_result = subprocess.run( # nosec B603
|
|
[tea_bin, "login", "add", "--name", login_name, "--url", gitea_url, "--token", token],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
if add_result.returncode != 0:
|
|
raise TeaCLIError(
|
|
f"tea login add failed (rc={add_result.returncode})\n"
|
|
f"stdout: {add_result.stdout.strip()}\n"
|
|
f"stderr: {add_result.stderr.strip()}"
|
|
)
|
|
|
|
default_result = subprocess.run( # nosec B603
|
|
[tea_bin, "login", "default", login_name],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
if default_result.returncode != 0:
|
|
raise TeaCLIError(
|
|
f"tea login default failed (rc={default_result.returncode})\n"
|
|
f"stdout: {default_result.stdout.strip()}\n"
|
|
f"stderr: {default_result.stderr.strip()}"
|
|
)
|
|
|
|
|
|
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.
|
|
|
|
Retries up to ``MAX_RETRIES`` times on transient HTTP errors
|
|
(502/503/504/429) detected in stderr/stdout, with exponential
|
|
backoff. Non-transient errors fail immediately.
|
|
|
|
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 after retries are exhausted.
|
|
"""
|
|
cmd = [self._tea, *args]
|
|
if json_output:
|
|
cmd.extend(["--output", "json"])
|
|
|
|
def _execute() -> str:
|
|
try:
|
|
result = subprocess.run( # nosec B603
|
|
cmd,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
except FileNotFoundError as e:
|
|
raise TeaCLIError(f"tea binary not found ('{self._tea}'). Install tea or add it to PATH.") from e
|
|
if result.returncode != 0:
|
|
parts = [
|
|
f"tea command failed (rc={result.returncode}): {' '.join(args)}",
|
|
f"stdout: {result.stdout.strip()}" if result.stdout.strip() else "",
|
|
f"stderr: {result.stderr.strip()}" if result.stderr.strip() else "",
|
|
]
|
|
msg = "\n".join(p for p in parts if p)
|
|
combined = f"{result.stdout} {result.stderr}".lower()
|
|
if any(str(code) in combined for code in RETRY_STATUS_CODES):
|
|
raise _TransientTeaError(msg)
|
|
raise TeaCLIError(msg)
|
|
return result.stdout.strip()
|
|
|
|
retry_decorator = retry(
|
|
stop=stop_after_attempt(MAX_RETRIES),
|
|
wait=wait_exponential(
|
|
multiplier=RETRY_BACKOFF_BASE,
|
|
min=RETRY_BACKOFF_BASE,
|
|
max=RETRY_BACKOFF_BASE**MAX_RETRIES,
|
|
),
|
|
retry=retry_if_exception_type(_TransientTeaError),
|
|
before_sleep=before_sleep_log(logger, logging.WARNING),
|
|
reraise=True,
|
|
)
|
|
try:
|
|
return retry_decorator(_execute)()
|
|
except _TransientTeaError as e:
|
|
raise TeaCLIError(str(e)) from e
|
|
|
|
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, "--description", 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)
|