528 lines
16 KiB
Python
528 lines
16 KiB
Python
"""
|
|
Gitea API Client for Allegro-Primus
|
|
Handles authentication, issues, PRs, and comments via Gitea API.
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import logging
|
|
from typing import Optional, List, Dict, Any
|
|
from dataclasses import dataclass, asdict
|
|
from enum import Enum
|
|
import requests
|
|
from urllib.parse import urljoin
|
|
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class IssueState(str, Enum):
|
|
OPEN = "open"
|
|
CLOSED = "closed"
|
|
ALL = "all"
|
|
|
|
|
|
class IssuePriority(str, Enum):
|
|
P0 = "P0" # Critical
|
|
P1 = "P1" # High
|
|
P2 = "P2" # Medium
|
|
P3 = "P3" # Low
|
|
|
|
|
|
@dataclass
|
|
class Issue:
|
|
id: int
|
|
number: int
|
|
title: str
|
|
body: str
|
|
state: str
|
|
user: Dict[str, Any]
|
|
labels: List[Dict[str, Any]]
|
|
assignee: Optional[Dict[str, Any]]
|
|
assignees: List[Dict[str, Any]]
|
|
milestone: Optional[Dict[str, Any]]
|
|
created_at: str
|
|
updated_at: str
|
|
closed_at: Optional[str]
|
|
due_date: Optional[str]
|
|
html_url: str
|
|
|
|
@property
|
|
def priority(self) -> Optional[str]:
|
|
"""Extract priority from labels."""
|
|
for label in self.labels:
|
|
name = label.get("name", "")
|
|
if name.startswith("P") and name[1:].isdigit():
|
|
return name
|
|
return None
|
|
|
|
|
|
@dataclass
|
|
class PullRequest:
|
|
id: int
|
|
number: int
|
|
title: str
|
|
body: str
|
|
state: str
|
|
user: Dict[str, Any]
|
|
head: Dict[str, Any]
|
|
base: Dict[str, Any]
|
|
mergeable: Optional[bool]
|
|
merged: bool
|
|
merged_at: Optional[str]
|
|
created_at: str
|
|
updated_at: str
|
|
html_url: str
|
|
|
|
|
|
class GiteaClient:
|
|
"""Client for interacting with Gitea API."""
|
|
|
|
def __init__(self, base_url: Optional[str] = None, token: Optional[str] = None):
|
|
"""
|
|
Initialize Gitea client.
|
|
|
|
Args:
|
|
base_url: Gitea instance URL (defaults to GITEA_URL env var)
|
|
token: API token (defaults to GITEA_TOKEN env var)
|
|
"""
|
|
self.base_url = (base_url or os.getenv("GITEA_URL", "")).rstrip("/")
|
|
self.token = token or os.getenv("GITEA_TOKEN", "")
|
|
|
|
if not self.base_url:
|
|
raise ValueError("Gitea base URL required. Set GITEA_URL env var or pass to constructor.")
|
|
if not self.token:
|
|
raise ValueError("Gitea token required. Set GITEA_TOKEN env var or pass to constructor.")
|
|
|
|
self.api_url = urljoin(self.base_url + "/", "api/v1")
|
|
self.session = requests.Session()
|
|
self.session.headers.update({
|
|
"Authorization": f"token {self.token}",
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json"
|
|
})
|
|
|
|
logger.info(f"GiteaClient initialized for {self.base_url}")
|
|
|
|
def _request(self, method: str, endpoint: str, **kwargs) -> Any:
|
|
"""Make authenticated request to Gitea API."""
|
|
url = f"{self.api_url}/{endpoint.lstrip('/')}"
|
|
try:
|
|
response = self.session.request(method, url, **kwargs)
|
|
response.raise_for_status()
|
|
return response.json() if response.content else None
|
|
except requests.exceptions.RequestException as e:
|
|
logger.error(f"API request failed: {e}")
|
|
if hasattr(e.response, 'text'):
|
|
logger.error(f"Response: {e.response.text}")
|
|
raise
|
|
|
|
# ==================== Issue Operations ====================
|
|
|
|
def list_issues(
|
|
self,
|
|
owner: str,
|
|
repo: str,
|
|
state: IssueState = IssueState.OPEN,
|
|
labels: Optional[List[str]] = None,
|
|
assignee: Optional[str] = None,
|
|
milestone: Optional[str] = None,
|
|
limit: int = 50,
|
|
page: int = 1
|
|
) -> List[Issue]:
|
|
"""
|
|
List issues for a repository with optional filters.
|
|
|
|
Args:
|
|
owner: Repository owner
|
|
repo: Repository name
|
|
state: Filter by state (open/closed/all)
|
|
labels: Filter by label names
|
|
assignee: Filter by assignee username
|
|
milestone: Filter by milestone
|
|
limit: Results per page
|
|
page: Page number
|
|
"""
|
|
params = {
|
|
"state": state.value,
|
|
"limit": limit,
|
|
"page": page
|
|
}
|
|
if labels:
|
|
params["labels"] = ",".join(labels)
|
|
if assignee:
|
|
params["assignee"] = assignee
|
|
if milestone:
|
|
params["milestone"] = milestone
|
|
|
|
data = self._request("GET", f"/repos/{owner}/{repo}/issues", params=params)
|
|
return [self._parse_issue(item) for item in data]
|
|
|
|
def get_issue(self, owner: str, repo: str, issue_number: int) -> Issue:
|
|
"""Get detailed information about a specific issue."""
|
|
data = self._request("GET", f"/repos/{owner}/{repo}/issues/{issue_number}")
|
|
return self._parse_issue(data)
|
|
|
|
def create_issue(
|
|
self,
|
|
owner: str,
|
|
repo: str,
|
|
title: str,
|
|
body: str = "",
|
|
labels: Optional[List[str]] = None,
|
|
assignees: Optional[List[str]] = None,
|
|
milestone: Optional[int] = None
|
|
) -> Issue:
|
|
"""
|
|
Create a new issue.
|
|
|
|
Args:
|
|
owner: Repository owner
|
|
repo: Repository name
|
|
title: Issue title
|
|
body: Issue description
|
|
labels: List of label names
|
|
assignees: List of usernames to assign
|
|
milestone: Milestone number
|
|
"""
|
|
payload = {"title": title, "body": body}
|
|
if labels:
|
|
payload["labels"] = labels
|
|
if assignees:
|
|
payload["assignees"] = assignees
|
|
if milestone:
|
|
payload["milestone"] = milestone
|
|
|
|
data = self._request("POST", f"/repos/{owner}/{repo}/issues", json=payload)
|
|
logger.info(f"Created issue #{data['number']} in {owner}/{repo}")
|
|
return self._parse_issue(data)
|
|
|
|
def update_issue(
|
|
self,
|
|
owner: str,
|
|
repo: str,
|
|
issue_number: int,
|
|
title: Optional[str] = None,
|
|
body: Optional[str] = None,
|
|
state: Optional[str] = None,
|
|
labels: Optional[List[str]] = None,
|
|
assignees: Optional[List[str]] = None,
|
|
milestone: Optional[int] = None
|
|
) -> Issue:
|
|
"""Update an existing issue."""
|
|
payload = {}
|
|
if title is not None:
|
|
payload["title"] = title
|
|
if body is not None:
|
|
payload["body"] = body
|
|
if state is not None:
|
|
payload["state"] = state
|
|
if labels is not None:
|
|
payload["labels"] = labels
|
|
if assignees is not None:
|
|
payload["assignees"] = assignees
|
|
if milestone is not None:
|
|
payload["milestone"] = milestone
|
|
|
|
data = self._request("PATCH", f"/repos/{owner}/{repo}/issues/{issue_number}", json=payload)
|
|
logger.info(f"Updated issue #{issue_number} in {owner}/{repo}")
|
|
return self._parse_issue(data)
|
|
|
|
def close_issue(self, owner: str, repo: str, issue_number: int) -> Issue:
|
|
"""Close an issue."""
|
|
return self.update_issue(owner, repo, issue_number, state="closed")
|
|
|
|
def reopen_issue(self, owner: str, repo: str, issue_number: int) -> Issue:
|
|
"""Reopen a closed issue."""
|
|
return self.update_issue(owner, repo, issue_number, state="open")
|
|
|
|
def add_issue_comment(
|
|
self,
|
|
owner: str,
|
|
repo: str,
|
|
issue_number: int,
|
|
body: str
|
|
) -> Dict[str, Any]:
|
|
"""Add a comment to an issue."""
|
|
payload = {"body": body}
|
|
data = self._request(
|
|
"POST",
|
|
f"/repos/{owner}/{repo}/issues/{issue_number}/comments",
|
|
json=payload
|
|
)
|
|
logger.info(f"Added comment to issue #{issue_number}")
|
|
return data
|
|
|
|
def get_issue_comments(self, owner: str, repo: str, issue_number: int) -> List[Dict[str, Any]]:
|
|
"""Get all comments for an issue."""
|
|
return self._request("GET", f"/repos/{owner}/{repo}/issues/{issue_number}/comments")
|
|
|
|
def get_priority_issues(
|
|
self,
|
|
owner: str,
|
|
repo: str,
|
|
priorities: List[str] = None,
|
|
state: IssueState = IssueState.OPEN
|
|
) -> List[Issue]:
|
|
"""
|
|
Get issues filtered by priority labels.
|
|
|
|
Args:
|
|
owner: Repository owner
|
|
repo: Repository name
|
|
priorities: List of priority labels (e.g., ["P0", "P1"])
|
|
state: Issue state filter
|
|
"""
|
|
if priorities is None:
|
|
priorities = ["P0", "P1"]
|
|
|
|
all_issues = []
|
|
for priority in priorities:
|
|
issues = self.list_issues(owner, repo, state=state, labels=[priority])
|
|
all_issues.extend(issues)
|
|
|
|
# Sort by priority (P0 first) then by creation date
|
|
priority_order = {p: i for i, p in enumerate(["P0", "P1", "P2", "P3"])}
|
|
all_issues.sort(key=lambda x: (
|
|
priority_order.get(x.priority or "P3", 99),
|
|
x.created_at
|
|
))
|
|
|
|
return all_issues
|
|
|
|
# ==================== Pull Request Operations ====================
|
|
|
|
def list_pull_requests(
|
|
self,
|
|
owner: str,
|
|
repo: str,
|
|
state: str = "open",
|
|
sort: str = "created",
|
|
limit: int = 50
|
|
) -> List[PullRequest]:
|
|
"""
|
|
List pull requests for a repository.
|
|
|
|
Args:
|
|
owner: Repository owner
|
|
repo: Repository name
|
|
state: Filter by state (open/closed/all)
|
|
sort: Sort field (created/updated/popularity/long-running)
|
|
limit: Results per page
|
|
"""
|
|
params = {"state": state, "sort": sort, "limit": limit}
|
|
data = self._request("GET", f"/repos/{owner}/{repo}/pulls", params=params)
|
|
return [self._parse_pr(item) for item in data]
|
|
|
|
def get_pull_request(self, owner: str, repo: str, pr_number: int) -> PullRequest:
|
|
"""Get detailed information about a specific PR."""
|
|
data = self._request("GET", f"/repos/{owner}/{repo}/pulls/{pr_number}")
|
|
return self._parse_pr(data)
|
|
|
|
def create_pull_request(
|
|
self,
|
|
owner: str,
|
|
repo: str,
|
|
title: str,
|
|
head: str,
|
|
base: str,
|
|
body: str = "",
|
|
draft: bool = False
|
|
) -> PullRequest:
|
|
"""
|
|
Create a new pull request.
|
|
|
|
Args:
|
|
owner: Repository owner
|
|
repo: Repository name
|
|
title: PR title
|
|
head: Branch containing changes
|
|
base: Branch to merge into
|
|
body: PR description
|
|
draft: Create as draft PR
|
|
"""
|
|
payload = {
|
|
"title": title,
|
|
"head": head,
|
|
"base": base,
|
|
"body": body,
|
|
"draft": draft
|
|
}
|
|
data = self._request("POST", f"/repos/{owner}/{repo}/pulls", json=payload)
|
|
logger.info(f"Created PR #{data['number']} in {owner}/{repo}")
|
|
return self._parse_pr(data)
|
|
|
|
def merge_pull_request(
|
|
self,
|
|
owner: str,
|
|
repo: str,
|
|
pr_number: int,
|
|
merge_method: str = "merge",
|
|
title: Optional[str] = None,
|
|
message: Optional[str] = None,
|
|
delete_branch: bool = False
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Merge a pull request.
|
|
|
|
Args:
|
|
owner: Repository owner
|
|
repo: Repository name
|
|
pr_number: PR number
|
|
merge_method: merge/rebase/squash
|
|
title: Custom merge commit title
|
|
message: Custom merge commit message
|
|
delete_branch: Delete head branch after merge
|
|
"""
|
|
payload = {"Do": merge_method}
|
|
if title:
|
|
payload["MergeTitleField"] = title
|
|
if message:
|
|
payload["MergeMessageField"] = message
|
|
if delete_branch:
|
|
payload["delete_branch_after_merge"] = True
|
|
|
|
data = self._request(
|
|
"POST",
|
|
f"/repos/{owner}/{repo}/pulls/{pr_number}/merge",
|
|
json=payload
|
|
)
|
|
logger.info(f"Merged PR #{pr_number} in {owner}/{repo}")
|
|
return data
|
|
|
|
def update_pull_request(
|
|
self,
|
|
owner: str,
|
|
repo: str,
|
|
pr_number: int,
|
|
title: Optional[str] = None,
|
|
body: Optional[str] = None,
|
|
state: Optional[str] = None
|
|
) -> PullRequest:
|
|
"""Update a pull request."""
|
|
payload = {}
|
|
if title is not None:
|
|
payload["title"] = title
|
|
if body is not None:
|
|
payload["body"] = body
|
|
if state is not None:
|
|
payload["state"] = state
|
|
|
|
data = self._request("PATCH", f"/repos/{owner}/{repo}/pulls/{pr_number}", json=payload)
|
|
return self._parse_pr(data)
|
|
|
|
def add_pr_comment(
|
|
self,
|
|
owner: str,
|
|
repo: str,
|
|
pr_number: int,
|
|
body: str,
|
|
path: Optional[str] = None,
|
|
position: Optional[int] = None
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Add a comment to a PR.
|
|
|
|
Args:
|
|
owner: Repository owner
|
|
repo: Repository name
|
|
pr_number: PR number
|
|
body: Comment text
|
|
path: File path (for line comments)
|
|
position: Line number (for line comments)
|
|
"""
|
|
payload = {"body": body}
|
|
if path and position:
|
|
payload["path"] = path
|
|
payload["position"] = position
|
|
|
|
data = self._request(
|
|
"POST",
|
|
f"/repos/{owner}/{repo}/pulls/{pr_number}/comments",
|
|
json=payload
|
|
)
|
|
return data
|
|
|
|
def get_pr_comments(self, owner: str, repo: str, pr_number: int) -> List[Dict[str, Any]]:
|
|
"""Get all comments for a PR."""
|
|
return self._request("GET", f"/repos/{owner}/{repo}/pulls/{pr_number}/comments")
|
|
|
|
def is_pr_mergeable(self, owner: str, repo: str, pr_number: int) -> bool:
|
|
"""Check if a PR is mergeable."""
|
|
pr = self.get_pull_request(owner, repo, pr_number)
|
|
return pr.mergeable is True
|
|
|
|
# ==================== Utility Methods ====================
|
|
|
|
def get_user(self) -> Dict[str, Any]:
|
|
"""Get current authenticated user info."""
|
|
return self._request("GET", "/user")
|
|
|
|
def get_repos(self, limit: int = 50) -> List[Dict[str, Any]]:
|
|
"""List repositories accessible to the user."""
|
|
return self._request("GET", "/user/repos", params={"limit": limit})
|
|
|
|
def get_repo(self, owner: str, repo: str) -> Dict[str, Any]:
|
|
"""Get repository information."""
|
|
return self._request("GET", f"/repos/{owner}/{repo}")
|
|
|
|
# ==================== Helper Methods ====================
|
|
|
|
def _parse_issue(self, data: Dict[str, Any]) -> Issue:
|
|
"""Parse API response into Issue dataclass."""
|
|
return Issue(
|
|
id=data.get("id", 0),
|
|
number=data.get("number", 0),
|
|
title=data.get("title", ""),
|
|
body=data.get("body", ""),
|
|
state=data.get("state", ""),
|
|
user=data.get("user", {}),
|
|
labels=data.get("labels", []),
|
|
assignee=data.get("assignee"),
|
|
assignees=data.get("assignees", []),
|
|
milestone=data.get("milestone"),
|
|
created_at=data.get("created_at", ""),
|
|
updated_at=data.get("updated_at", ""),
|
|
closed_at=data.get("closed_at"),
|
|
due_date=data.get("due_date"),
|
|
html_url=data.get("html_url", "")
|
|
)
|
|
|
|
def _parse_pr(self, data: Dict[str, Any]) -> PullRequest:
|
|
"""Parse API response into PullRequest dataclass."""
|
|
return PullRequest(
|
|
id=data.get("id", 0),
|
|
number=data.get("number", 0),
|
|
title=data.get("title", ""),
|
|
body=data.get("body", ""),
|
|
state=data.get("state", ""),
|
|
user=data.get("user", {}),
|
|
head=data.get("head", {}),
|
|
base=data.get("base", {}),
|
|
mergeable=data.get("mergeable"),
|
|
merged=data.get("merged", False),
|
|
merged_at=data.get("merged_at"),
|
|
created_at=data.get("created_at", ""),
|
|
updated_at=data.get("updated_at", ""),
|
|
html_url=data.get("html_url", "")
|
|
)
|
|
|
|
|
|
# ==================== Example Usage ====================
|
|
|
|
if __name__ == "__main__":
|
|
import os
|
|
|
|
# Load from environment
|
|
client = GiteaClient()
|
|
|
|
# Example: Get user info
|
|
user = client.get_user()
|
|
print(f"Authenticated as: {user.get('login')}")
|
|
|
|
# Example: List repositories
|
|
repos = client.get_repos(limit=10)
|
|
print(f"\nFound {len(repos)} repositories:")
|
|
for repo in repos:
|
|
print(f" - {repo['full_name']}")
|