meeks
21eefd9824
- IssueTools.get_issue now returns IssueModel instead of JSON string - PRTools.get_pull_request now returns PullRequestModel instead of JSON string - PRTools.create_pull_request now returns PullRequestModel instead of JSON string - PRTools.update_pull_request now returns PullRequestModel instead of JSON string - All methods have proper return type hints and raise exceptions on error - Updated tests to verify model objects are returned directly - Marked issue 5.4 as resolved in bad_code.md
222 lines
7.8 KiB
Python
222 lines
7.8 KiB
Python
"""Tools for Gitea pull request operations."""
|
||
|
||
import json
|
||
import logging
|
||
from typing import Any
|
||
from gitea.client import GiteaClient
|
||
from gitea.models import PullRequestModel, CommentModel
|
||
|
||
logger: logging.Logger = logging.getLogger("gitea.tools.pr_tools")
|
||
|
||
|
||
_MAX_DIFF_CHARS: int = 15_000
|
||
|
||
|
||
def _truncate_diff(
|
||
text: str,
|
||
max_chars: int = _MAX_DIFF_CHARS,
|
||
char_offset: int = 0,
|
||
) -> str:
|
||
"""Slice [char_offset : char_offset+max_chars] from text, cutting at a hunk boundary."""
|
||
total: int = len(text)
|
||
chunk: str = text[char_offset : char_offset + max_chars]
|
||
if char_offset > 0 or (char_offset + max_chars) < total:
|
||
# Try to cut at a diff hunk boundary (@@) for coherence
|
||
hunk_boundary: int = chunk.rfind("\n@@")
|
||
if hunk_boundary > int(len(chunk) * 0.6):
|
||
chunk = chunk[:hunk_boundary]
|
||
next_offset: int = char_offset + len(chunk)
|
||
chunk += (
|
||
f"\n\n[Diff truncated — {total} chars total. "
|
||
f"Showing chars {char_offset}–{next_offset}. "
|
||
f"Re-call with char_offset={next_offset} to read more.]"
|
||
)
|
||
return chunk
|
||
|
||
|
||
class PRTools:
|
||
"""Tools for Gitea pull request operations."""
|
||
|
||
def __init__(self, client: GiteaClient) -> None:
|
||
self._client = client
|
||
|
||
def get_pull_request(self, owner: str, repo: str, pull_number: int) -> PullRequestModel:
|
||
try:
|
||
return self._client.prs.get_pull_request(owner, repo, pull_number)
|
||
except Exception as e:
|
||
logger.error(f"Error getting pull request #{pull_number}: {e}", exc_info=True)
|
||
raise
|
||
|
||
def close_pull_request(self, owner: str, repo: str, pull_number: int) -> str:
|
||
try:
|
||
self._client.prs.close_pull_request(owner, repo, pull_number)
|
||
return f"Pull request #{pull_number} closed successfully."
|
||
except Exception as e:
|
||
return f"Error closing pull request: {str(e)}"
|
||
|
||
def get_pull_request_comments(
|
||
self,
|
||
owner: str,
|
||
repo: str,
|
||
pull_number: int,
|
||
limit: int = 20,
|
||
offset: int = 0,
|
||
) -> str:
|
||
"""Get comments on a pull request with optional paging.
|
||
|
||
Args:
|
||
limit: Maximum number of comments to return (default 20).
|
||
offset: Zero-based comment index to start from (default 0).
|
||
"""
|
||
try:
|
||
comments: list[CommentModel] = self._client.prs.get_pull_request_comments(
|
||
owner, repo, pull_number
|
||
)
|
||
total: int = len(comments)
|
||
page: list[CommentModel] = comments[offset : offset + limit]
|
||
result: str = json.dumps([c.model_dump() for c in page], indent=2)
|
||
if total > offset + limit:
|
||
next_offset: int = offset + limit
|
||
result += (
|
||
f"\n\n[{total} comments total — showing {offset}–{offset + len(page)}. "
|
||
f"Re-call with offset={next_offset} to see more.]"
|
||
)
|
||
return result
|
||
except Exception as e:
|
||
return f"Error getting PR comments: {str(e)}"
|
||
|
||
def list_assigned_pull_requests(self) -> list[dict[str, Any]]:
|
||
try:
|
||
repos = self._client.repos.list_all_user_repos()
|
||
all_prs: list[dict[str, Any]] = []
|
||
for repo_info in repos:
|
||
repo_owner = repo_info.owner
|
||
repo_name = repo_info.name
|
||
prs = self._client.prs.list_assigned_pull_requests(
|
||
repo_owner, repo_name
|
||
)
|
||
if prs:
|
||
all_prs.extend(
|
||
[
|
||
pr.model_dump() if hasattr(pr, "model_dump") else pr
|
||
for pr in prs
|
||
]
|
||
)
|
||
return all_prs
|
||
except Exception as e:
|
||
logger.error(f"Error listing assigned pull requests: {e}", exc_info=True)
|
||
return []
|
||
|
||
def list_pull_requests(self, owner: str, repo: str, state: str = "open") -> str:
|
||
try:
|
||
prs = self._client.prs.list_repo_pull_requests(owner, repo, state)
|
||
if not prs:
|
||
return f"No PRs in {owner}/{repo}."
|
||
summary = [f"#{pr.number}: {pr.title}" for pr in prs]
|
||
return "\n".join(summary)
|
||
except Exception as e:
|
||
return f"Error listing PRs: {str(e)}"
|
||
|
||
def create_pull_request(
|
||
self,
|
||
owner: str,
|
||
repo: str,
|
||
head: str,
|
||
base: str,
|
||
title: str,
|
||
description: str = "",
|
||
) -> PullRequestModel:
|
||
try:
|
||
return self._client.prs.create_pr_via_tea(
|
||
owner, repo, title, description, head, base
|
||
)
|
||
except Exception as e:
|
||
logger.error(f"Error creating PR in {owner}/{repo}: {e}", exc_info=True)
|
||
raise
|
||
|
||
def update_pull_request(
|
||
self,
|
||
owner: str,
|
||
repo: str,
|
||
pull_number: int,
|
||
title: str | None = None,
|
||
body: str | None = None,
|
||
state: str | None = None,
|
||
) -> PullRequestModel:
|
||
try:
|
||
return self._client.prs.update_pull_request(
|
||
owner, repo, pull_number, title, body, state
|
||
)
|
||
except Exception as e:
|
||
logger.error(f"Error updating PR #{pull_number}: {e}", exc_info=True)
|
||
raise
|
||
|
||
def add_label_to_pr(self, owner: str, repo: str, pr_number: int, label: str) -> str:
|
||
try:
|
||
self._client.prs.add_label_pr(owner, repo, pr_number, label)
|
||
return f"Label '{label}' added to PR #{pr_number}."
|
||
except Exception as e:
|
||
return f"Error adding label to PR #{pr_number}: {e}"
|
||
|
||
def get_pull_request_diff(
|
||
self,
|
||
owner: str,
|
||
repo: str,
|
||
pull_number: int,
|
||
max_chars: int = _MAX_DIFF_CHARS,
|
||
char_offset: int = 0,
|
||
) -> str:
|
||
"""Get the diff of a pull request, with truncation and offset paging.
|
||
|
||
Args:
|
||
max_chars: Maximum characters to return (default 15 000).
|
||
char_offset: Character offset to start reading from (default 0).
|
||
Increment by max_chars to page through a large diff.
|
||
"""
|
||
try:
|
||
diff: str = self._client.prs.get_pull_request_diff(owner, repo, pull_number)
|
||
return _truncate_diff(diff, max_chars, char_offset)
|
||
except Exception as e:
|
||
return f"Error getting PR diff: {str(e)}"
|
||
|
||
def get_pull_request_patch(
|
||
self,
|
||
owner: str,
|
||
repo: str,
|
||
pull_number: int,
|
||
max_chars: int = _MAX_DIFF_CHARS,
|
||
char_offset: int = 0,
|
||
) -> str:
|
||
"""Get the patch of a pull request, with truncation and offset paging.
|
||
|
||
Args:
|
||
max_chars: Maximum characters to return (default 15 000).
|
||
char_offset: Character offset to start reading from (default 0).
|
||
Increment by max_chars to page through a large patch.
|
||
"""
|
||
try:
|
||
patch: str = self._client.prs.get_pull_request_patch(
|
||
owner, repo, pull_number
|
||
)
|
||
return _truncate_diff(patch, max_chars, char_offset)
|
||
except Exception as e:
|
||
return f"Error getting PR patch: {str(e)}"
|
||
|
||
def approve_pull_request(
|
||
self, owner: str, repo: str, pull_number: int, comment: str
|
||
) -> str:
|
||
try:
|
||
self._client.prs.approve_pr(owner, repo, pull_number, comment)
|
||
return f"Approved PR #{pull_number}."
|
||
except Exception as e:
|
||
return f"Error approving PR: {str(e)}"
|
||
|
||
def request_changes(
|
||
self, owner: str, repo: str, pull_number: int, comment: str
|
||
) -> str:
|
||
try:
|
||
self._client.prs.request_changes_pr(owner, repo, pull_number, comment)
|
||
return f"Requested changes on PR #{pull_number}."
|
||
except Exception as e:
|
||
return f"Error requesting changes: {str(e)}"
|