coding-agent-gitea/gitea/prs_client.py

"""Pull Requests client for Gitea API operations."""

import logging
from dataclasses import dataclass
from typing import Callable

import httpx

from .models import (
    CommentModel,
    LabelModel,
    PullRequestFileModel,
    PullRequestModel,
    RepositoryModel,
    UserModel,
)

logger: logging.Logger = logging.getLogger("gitea.prs_client")


@dataclass
class ReviewRequest:
    event: str
    body: str


class PullRequestsClient:
    """HTTP client for Gitea Pull Requests API operations."""

    def __init__(
        self,
        base_url: str,
        client: httpx.Client,
        get_user: Callable[[], UserModel] | None = None,
        get_repos: Callable[[], list[RepositoryModel]] | None = None,
    ) -> None:
        """Initialize the PullRequestsClient.

        Args:
            base_url: The base URL for the Gitea API.
            client: The httpx client for making requests.
            get_user: Optional callable to get the authenticated user.
            get_repos: Optional callable to get all user repos.
        """
        self.base_url: str = base_url
        self.client: httpx.Client = client
        self._get_user: Callable[[], UserModel] | None = get_user
        self._get_repos: Callable[[], list[RepositoryModel]] | None = get_repos

    def list_repo_pull_requests(
        self, owner: str, repo: str, state: str = "open"
    ) -> list[PullRequestModel]:
        """List pull requests for a repository.

        Args:
            owner: Repository owner.
            repo: Repository name.
            state: PR state filter (open, closed, all).

        Returns:
            List of pull requests matching the criteria.
        """
        url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls?state={state}"
        response = self.client.get(url)
        if response.status_code == 404:
            return []
        response.raise_for_status()
        return [PullRequestModel(**item) for item in response.json()]

    def get_pull_request(
        self, owner: str, repo: str, pull_number: int
    ) -> PullRequestModel:
        """Get a specific pull request.

        Args:
            owner: Repository owner.
            repo: Repository name.
            pull_number: Pull request number.

        Returns:
            The requested pull request.
        """
        url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}"
        response = self.client.get(url)
        response.raise_for_status()
        return PullRequestModel(**response.json())

    def close_pull_request(
        self, owner: str, repo: str, pull_number: int
    ) -> PullRequestModel:
        """Close a pull request.

        Args:
            owner: Repository owner.
            repo: Repository name.
            pull_number: Pull request number.

        Returns:
            The updated pull request.
        """
        url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}"
        data: dict[str, str] = {"state": "closed"}
        response = self.client.patch(url, json=data)
        response.raise_for_status()
        return PullRequestModel(**response.json())

    def get_pull_request_comments(
        self, owner: str, repo: str, pull_number: int
    ) -> list[CommentModel]:
        """Get comments on a pull request.

        Args:
            owner: Repository owner.
            repo: Repository name.
            pull_number: Pull request number.

        Returns:
            List of comments on the pull request.
        """
        url = (
            f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{pull_number}/comments"
        )
        response = self.client.get(url)
        response.raise_for_status()
        return [CommentModel(**item) for item in response.json()]

    def get_pull_request_diff(self, owner: str, repo: str, pull_number: int) -> str:
        """Get the diff for a pull request.

        Args:
            owner: Repository owner.
            repo: Repository name.
            pull_number: Pull request number.

        Returns:
            The diff as a string.
        """
        url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}/diff"
        response = self.client.get(url)
        response.raise_for_status()
        return response.text

    def get_pull_request_patch(self, owner: str, repo: str, pull_number: int) -> str:
        """Get the patch for a pull request.

        Args:
            owner: Repository owner.
            repo: Repository name.
            pull_number: Pull request number.

        Returns:
            The patch as a string.
        """
        url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}/patch"
        response = self.client.get(url)
        response.raise_for_status()
        return response.text

    def get_pull_request_files(
        self, owner: str, repo: str, pull_number: int
    ) -> list[PullRequestFileModel]:
        """Get the files changed in a pull request.

        Args:
            owner: Repository owner.
            repo: Repository name.
            pull_number: Pull request number.

        Returns:
            List of files changed in the pull request.
        """
        url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}/files"
        response = self.client.get(url)
        response.raise_for_status()
        return [PullRequestFileModel(**item) for item in response.json()]

    def list_assigned_pull_requests(
        self, owner: str = "", repo: str = ""
    ) -> list[PullRequestModel]:
        """List all pull requests assigned to or authored by the authenticated user.

        Args:
            owner: Optional repository owner to filter by.
            repo: Optional repository name to filter by.

        Returns:
            List of pull requests assigned to or authored by the user.
        """
        try:
            if self._get_user is None or self._get_repos is None:
                logger.error("get_user and get_repos callables are required")
                return []

            user = self._get_user()
            if not user:
                return []
            username: str = user.login
            if owner and repo:
                response = self.client.get(
                    f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls?state=open",
                )
                if response.status_code == 404:
                    return []
                response.raise_for_status()
                all_prs: list[PullRequestModel] = [
                    PullRequestModel(**pr) for pr in response.json()
                ]
                return [
                    pr
                    for pr in all_prs
                    if (pr.assignee and pr.assignee.login == username)
                    or (pr.user and pr.user.login == username)
                ]
            all_prs: list[PullRequestModel] = []
            repos = self._get_repos()
            for r in repos:
                repo_owner = r.owner
                repo_name = r.name
                resp = self.client.get(
                    f"{self.base_url}/api/v1/repos/{repo_owner}/{repo_name}/pulls?state=open",
                )
                if resp.status_code == 200:
                    for pr_data in resp.json():
                        pr = PullRequestModel(**pr_data)
                        if (pr.assignee and pr.assignee.login == username) or (
                            pr.user and pr.user.login == username
                        ):
                            # Backfill repository if Gitea omitted it
                            if pr.repository is None:
                                pr = pr.model_copy(update={"repository": r})
                            all_prs.append(pr)
            return all_prs
        except Exception as e:
            logger.error(f"Error listing assigned pull requests: {e}", exc_info=True)
            return []

    def create_pull_request(
        self,
        owner: str,
        repo: str,
        head: str,
        base: str,
        title: str,
        description: str = "",
    ) -> PullRequestModel:
        """Create a new pull request.

        Args:
            owner: Repository owner.
            repo: Repository name.
            head: Head branch name.
            base: Base branch name.
            title: Pull request title.
            description: Pull request description.

        Returns:
            The created pull request.
        """
        try:
            url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls"
            data: dict[str, str] = {
                "title": title,
                "body": description,
                "head": head,
                "base": base,
            }
            response = self.client.post(url, json=data)
            response.raise_for_status()
            return PullRequestModel(**response.json())
        except Exception as e:
            logger.error(f"Error creating pull request: {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:
        """Update a pull request.

        Args:
            owner: Repository owner.
            repo: Repository name.
            pull_number: Pull request number.
            title: Optional new title.
            body: Optional new body.
            state: Optional new state.

        Returns:
            The updated pull request.
        """
        try:
            url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}"
            data: dict[str, str | None] = {}
            if title is not None:
                data["title"] = title
            if body is not None:
                data["body"] = body
            if state is not None:
                data["state"] = state
            response = self.client.patch(url, json=data)
            response.raise_for_status()
            return PullRequestModel(**response.json())
        except Exception as e:
            logger.error(f"Error updating pull request: {e}", exc_info=True)
            raise

    def create_pr_via_tea(
        self, owner: str, repo: str, title: str, description: str, head: str, base: str
    ) -> PullRequestModel:
        """Create a pull request (alias for create_pull_request).

        Args:
            owner: Repository owner.
            repo: Repository name.
            title: Pull request title.
            description: Pull request description.
            head: Head branch name.
            base: Base branch name.

        Returns:
            The created pull request.
        """
        return self.create_pull_request(owner, repo, head, base, title, description)

    def approve_pr(
        self, owner: str, repo: str, pr_number: int, comment: str
    ) -> dict[str, object]:
        """Approve a pull request.

        Args:
            owner: Repository owner.
            repo: Repository name.
            pr_number: Pull request number.
            comment: Review comment.

        Returns:
            The review response.
        """
        url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}/reviews"
        review: ReviewRequest = ReviewRequest(event="APPROVED", body=comment)
        response = self.client.post(url, json={"event": review.event, "body": review.body})
        response.raise_for_status()
        return response.json()

    def request_changes_pr(
        self, owner: str, repo: str, pr_number: int, comment: str
    ) -> dict[str, object]:
        """Request changes on a pull request.

        Args:
            owner: Repository owner.
            repo: Repository name.
            pr_number: Pull request number.
            comment: Review comment.

        Returns:
            The review response.
        """
        url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}/reviews"
        review: ReviewRequest = ReviewRequest(event="REQUEST_CHANGES", body=comment)
        response = self.client.post(url, json={"event": review.event, "body": review.body})
        response.raise_for_status()
        return response.json()

    def get_pr_reviews(
        self, owner: str, repo: str, pr_number: int
    ) -> list[dict[str, object]]:
        """Get reviews for a pull request.

        Args:
            owner: Repository owner.
            repo: Repository name.
            pr_number: Pull request number.

        Returns:
            List of reviews.
        """
        url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}/reviews"
        response = self.client.get(url)
        if response.status_code == 404:
            return []
        response.raise_for_status()
        return response.json()

    def dismiss_review_pr(
        self, owner: str, repo: str, pr_number: int, review_id: int, message: str
    ) -> dict[str, object]:
        """Dismiss a review on a pull request.

        Args:
            owner: Repository owner.
            repo: Repository name.
            pr_number: Pull request number.
            review_id: Review ID to dismiss.
            message: Dismissal message.

        Returns:
            The dismissal response.
        """
        url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}/reviews/{review_id}/dismissals"
        data: dict[str, str] = {"message": message}
        response = self.client.post(url, json=data)
        response.raise_for_status()
        return response.json()

    def add_label_pr(
        self, owner: str, repo: str, pr_number: int, label: str
    ) -> LabelModel:
        """Add a label to a pull request.

        Args:
            owner: Repository owner.
            repo: Repository name.
            pr_number: Pull request number.
            label: Label name or ID.

        Returns:
            The added label.
        """
        url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{pr_number}/labels"
        data: list[str] = [label]
        response = self.client.post(url, json=data)
        response.raise_for_status()
        return LabelModel(**response.json())

    def merge_pull_request(
        self,
        owner: str,
        repo: str,
        pull_number: int,
        style: str = "squash",
        title: str = "",
        message: str = "",
    ) -> bool:
        """Merge a pull request.

        Args:
            owner: Repository owner.
            repo: Repository name.
            pull_number: Pull request number.
            style: Merge style (squash, merge, rebase).
            title: Optional merge commit title.
            message: Optional merge commit message.

        Returns:
            True if merge was successful.
        """
        try:
            url = (
                f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}/merge"
            )
            data: dict[str, str] = {
                "Do": style,
                "MergeTitleField": title,
                "MergeMessageField": message,
            }
            response = self.client.post(url, json=data)
            response.raise_for_status()
            return True
        except Exception as e:
            logger.error(
                f"Error merging pull request {pull_number}: {e}", exc_info=True
            )
            raise