448 lines
11 KiB
Python
448 lines
11 KiB
Python
"""Gitea API Client
|
|
|
|
A unified client for interacting with the Gitea REST API.
|
|
Provides methods for issues, pull requests, comments, and repository operations.
|
|
"""
|
|
|
|
import os
|
|
from typing import Any
|
|
|
|
import requests
|
|
|
|
|
|
class GiteaClient:
|
|
"""Client for Gitea API operations."""
|
|
|
|
def __init__(
|
|
self,
|
|
api_url: str | None = None,
|
|
token: str | None = None,
|
|
timeout: int = 30,
|
|
):
|
|
"""Initialize the Gitea client.
|
|
|
|
Args:
|
|
api_url: Gitea API base URL. Defaults to AI_REVIEW_API_URL env var.
|
|
token: API token. Defaults to AI_REVIEW_TOKEN env var.
|
|
timeout: Request timeout in seconds.
|
|
"""
|
|
self.api_url = api_url or os.environ.get("AI_REVIEW_API_URL", "")
|
|
self.token = token or os.environ.get("AI_REVIEW_TOKEN", "")
|
|
self.timeout = timeout
|
|
|
|
if not self.api_url:
|
|
raise ValueError("Gitea API URL is required")
|
|
if not self.token:
|
|
raise ValueError("Gitea API token is required")
|
|
|
|
self.headers = {
|
|
"Authorization": f"token {self.token}",
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json",
|
|
}
|
|
|
|
def _request(
|
|
self,
|
|
method: str,
|
|
endpoint: str,
|
|
json: dict | None = None,
|
|
params: dict | None = None,
|
|
) -> dict | list:
|
|
"""Make an API request.
|
|
|
|
Args:
|
|
method: HTTP method (GET, POST, PATCH, DELETE).
|
|
endpoint: API endpoint (without base URL).
|
|
json: Request body for POST/PATCH.
|
|
params: Query parameters.
|
|
|
|
Returns:
|
|
Response JSON data.
|
|
|
|
Raises:
|
|
requests.HTTPError: If the request fails.
|
|
"""
|
|
url = f"{self.api_url}{endpoint}"
|
|
response = requests.request(
|
|
method=method,
|
|
url=url,
|
|
headers=self.headers,
|
|
json=json,
|
|
params=params,
|
|
timeout=self.timeout,
|
|
)
|
|
response.raise_for_status()
|
|
|
|
if response.status_code == 204:
|
|
return {}
|
|
return response.json()
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Issue Operations
|
|
# -------------------------------------------------------------------------
|
|
|
|
def create_issue(
|
|
self,
|
|
owner: str,
|
|
repo: str,
|
|
title: str,
|
|
body: str,
|
|
labels: list[int] | None = None,
|
|
) -> dict:
|
|
"""Create a new issue.
|
|
|
|
Args:
|
|
owner: Repository owner.
|
|
repo: Repository name.
|
|
title: Issue title.
|
|
body: Issue body.
|
|
labels: Optional list of label IDs.
|
|
|
|
Returns:
|
|
Created issue object.
|
|
"""
|
|
payload = {
|
|
"title": title,
|
|
"body": body,
|
|
}
|
|
if labels:
|
|
payload["labels"] = labels
|
|
|
|
return self._request(
|
|
"POST",
|
|
f"/repos/{owner}/{repo}/issues",
|
|
json=payload,
|
|
)
|
|
|
|
def update_issue(
|
|
self,
|
|
owner: str,
|
|
repo: str,
|
|
index: int,
|
|
title: str | None = None,
|
|
body: str | None = None,
|
|
state: str | None = None,
|
|
) -> dict:
|
|
"""Update an existing issue.
|
|
|
|
Args:
|
|
owner: Repository owner.
|
|
repo: Repository name.
|
|
index: Issue number.
|
|
title: New title.
|
|
body: New body.
|
|
state: New state (open, closed).
|
|
|
|
Returns:
|
|
Updated issue object.
|
|
"""
|
|
payload = {}
|
|
if title:
|
|
payload["title"] = title
|
|
if body:
|
|
payload["body"] = body
|
|
if state:
|
|
payload["state"] = state
|
|
|
|
return self._request(
|
|
"PATCH",
|
|
f"/repos/{owner}/{repo}/issues/{index}",
|
|
json=payload,
|
|
)
|
|
|
|
def list_issues(
|
|
self,
|
|
owner: str,
|
|
repo: str,
|
|
state: str = "open",
|
|
labels: list[str] | None = None,
|
|
page: int = 1,
|
|
limit: int = 30,
|
|
) -> list[dict]:
|
|
"""List issues in a repository.
|
|
|
|
Args:
|
|
owner: Repository owner.
|
|
repo: Repository name.
|
|
state: Issue state (open, closed, all).
|
|
labels: Filter by labels.
|
|
page: Page number.
|
|
limit: Items per page.
|
|
|
|
Returns:
|
|
List of issue objects.
|
|
"""
|
|
params = {
|
|
"state": state,
|
|
"page": page,
|
|
"limit": limit,
|
|
}
|
|
if labels:
|
|
params["labels"] = ",".join(labels)
|
|
|
|
return self._request("GET", f"/repos/{owner}/{repo}/issues", params=params)
|
|
|
|
def get_issue(self, owner: str, repo: str, index: int) -> dict:
|
|
"""Get a single issue.
|
|
|
|
Args:
|
|
owner: Repository owner.
|
|
repo: Repository name.
|
|
index: Issue number.
|
|
|
|
Returns:
|
|
Issue object.
|
|
"""
|
|
return self._request("GET", f"/repos/{owner}/{repo}/issues/{index}")
|
|
|
|
def create_issue_comment(
|
|
self,
|
|
owner: str,
|
|
repo: str,
|
|
index: int,
|
|
body: str,
|
|
) -> dict:
|
|
"""Create a comment on an issue.
|
|
|
|
Args:
|
|
owner: Repository owner.
|
|
repo: Repository name.
|
|
index: Issue number.
|
|
body: Comment body.
|
|
|
|
Returns:
|
|
Created comment object.
|
|
"""
|
|
return self._request(
|
|
"POST",
|
|
f"/repos/{owner}/{repo}/issues/{index}/comments",
|
|
json={"body": body},
|
|
)
|
|
|
|
def update_issue_comment(
|
|
self,
|
|
owner: str,
|
|
repo: str,
|
|
comment_id: int,
|
|
body: str,
|
|
) -> dict:
|
|
"""Update an existing comment.
|
|
|
|
Args:
|
|
owner: Repository owner.
|
|
repo: Repository name.
|
|
comment_id: Comment ID.
|
|
body: Updated comment body.
|
|
|
|
Returns:
|
|
Updated comment object.
|
|
"""
|
|
return self._request(
|
|
"PATCH",
|
|
f"/repos/{owner}/{repo}/issues/comments/{comment_id}",
|
|
json={"body": body},
|
|
)
|
|
|
|
def list_issue_comments(
|
|
self,
|
|
owner: str,
|
|
repo: str,
|
|
index: int,
|
|
) -> list[dict]:
|
|
"""List comments on an issue.
|
|
|
|
Args:
|
|
owner: Repository owner.
|
|
repo: Repository name.
|
|
index: Issue number.
|
|
|
|
Returns:
|
|
List of comment objects.
|
|
"""
|
|
return self._request("GET", f"/repos/{owner}/{repo}/issues/{index}/comments")
|
|
|
|
def add_issue_labels(
|
|
self,
|
|
owner: str,
|
|
repo: str,
|
|
index: int,
|
|
labels: list[int],
|
|
) -> list[dict]:
|
|
"""Add labels to an issue.
|
|
|
|
Args:
|
|
owner: Repository owner.
|
|
repo: Repository name.
|
|
index: Issue number.
|
|
labels: List of label IDs to add.
|
|
|
|
Returns:
|
|
List of label objects.
|
|
"""
|
|
return self._request(
|
|
"POST",
|
|
f"/repos/{owner}/{repo}/issues/{index}/labels",
|
|
json={"labels": labels},
|
|
)
|
|
|
|
def get_repo_labels(self, owner: str, repo: str) -> list[dict]:
|
|
"""Get all labels for a repository.
|
|
|
|
Args:
|
|
owner: Repository owner.
|
|
repo: Repository name.
|
|
|
|
Returns:
|
|
List of label objects.
|
|
"""
|
|
return self._request("GET", f"/repos/{owner}/{repo}/labels")
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Pull Request Operations
|
|
# -------------------------------------------------------------------------
|
|
|
|
def get_pull_request(self, owner: str, repo: str, index: int) -> dict:
|
|
"""Get a pull request.
|
|
|
|
Args:
|
|
owner: Repository owner.
|
|
repo: Repository name.
|
|
index: PR number.
|
|
|
|
Returns:
|
|
Pull request object.
|
|
"""
|
|
return self._request("GET", f"/repos/{owner}/{repo}/pulls/{index}")
|
|
|
|
def get_pull_request_diff(self, owner: str, repo: str, index: int) -> str:
|
|
"""Get the diff for a pull request.
|
|
|
|
Args:
|
|
owner: Repository owner.
|
|
repo: Repository name.
|
|
index: PR number.
|
|
|
|
Returns:
|
|
Diff text.
|
|
"""
|
|
url = f"{self.api_url}/repos/{owner}/{repo}/pulls/{index}.diff"
|
|
response = requests.get(
|
|
url,
|
|
headers={
|
|
"Authorization": f"token {self.token}",
|
|
"Accept": "text/plain",
|
|
},
|
|
timeout=self.timeout,
|
|
)
|
|
response.raise_for_status()
|
|
return response.text
|
|
|
|
def list_pull_request_files(
|
|
self,
|
|
owner: str,
|
|
repo: str,
|
|
index: int,
|
|
) -> list[dict]:
|
|
"""List files changed in a pull request.
|
|
|
|
Args:
|
|
owner: Repository owner.
|
|
repo: Repository name.
|
|
index: PR number.
|
|
|
|
Returns:
|
|
List of changed file objects.
|
|
"""
|
|
return self._request("GET", f"/repos/{owner}/{repo}/pulls/{index}/files")
|
|
|
|
def create_pull_request_review(
|
|
self,
|
|
owner: str,
|
|
repo: str,
|
|
index: int,
|
|
body: str,
|
|
event: str = "COMMENT",
|
|
comments: list[dict] | None = None,
|
|
) -> dict:
|
|
"""Create a review on a pull request.
|
|
|
|
Args:
|
|
owner: Repository owner.
|
|
repo: Repository name.
|
|
index: PR number.
|
|
body: Review body.
|
|
event: Review event (APPROVE, REQUEST_CHANGES, COMMENT).
|
|
comments: List of inline comments.
|
|
|
|
Returns:
|
|
Created review object.
|
|
"""
|
|
payload: dict[str, Any] = {
|
|
"body": body,
|
|
"event": event,
|
|
}
|
|
if comments:
|
|
payload["comments"] = comments
|
|
|
|
return self._request(
|
|
"POST",
|
|
f"/repos/{owner}/{repo}/pulls/{index}/reviews",
|
|
json=payload,
|
|
)
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Repository Operations
|
|
# -------------------------------------------------------------------------
|
|
|
|
def get_repository(self, owner: str, repo: str) -> dict:
|
|
"""Get repository information.
|
|
|
|
Args:
|
|
owner: Repository owner.
|
|
repo: Repository name.
|
|
|
|
Returns:
|
|
Repository object.
|
|
"""
|
|
return self._request("GET", f"/repos/{owner}/{repo}")
|
|
|
|
def get_file_contents(
|
|
self,
|
|
owner: str,
|
|
repo: str,
|
|
filepath: str,
|
|
ref: str | None = None,
|
|
) -> dict:
|
|
"""Get file contents from a repository.
|
|
|
|
Args:
|
|
owner: Repository owner.
|
|
repo: Repository name.
|
|
filepath: Path to file.
|
|
ref: Git ref (branch, tag, commit).
|
|
|
|
Returns:
|
|
File content object with base64-encoded content.
|
|
"""
|
|
params = {}
|
|
if ref:
|
|
params["ref"] = ref
|
|
return self._request(
|
|
"GET",
|
|
f"/repos/{owner}/{repo}/contents/{filepath}",
|
|
params=params,
|
|
)
|
|
|
|
def get_branch(self, owner: str, repo: str, branch: str) -> dict:
|
|
"""Get branch information.
|
|
|
|
Args:
|
|
owner: Repository owner.
|
|
repo: Repository name.
|
|
branch: Branch name.
|
|
|
|
Returns:
|
|
Branch object.
|
|
"""
|
|
return self._request("GET", f"/repos/{owner}/{repo}/branches/{branch}")
|