aider-gitea/aider_gitea/gitea_client.py

import logging
from collections.abc import Iterator

import requests

logger = logging.getLogger(__name__)


class GiteaClient:
    """Client for interacting with the Gitea API.

    This class provides methods to interact with a Gitea instance's API,
    including retrieving repository information, creating branches, and fetching issues.

    Read more about the Gitea API here: https://gitea.com/api/swagger

    Attributes:
        gitea_url (str): The base URL for the Gitea API endpoints.
        session (requests.Session): HTTP session for making API requests.
    """

    def __init__(self, gitea_url: str, token: str) -> None:
        """Initialize a new Gitea API client.

        Args:
            gitea_url (str): Base URL for the Gitea instance (without '/api/v1').
            token (str): Authentication token for the Gitea API. If empty, requests will be unauthenticated.

        Raises:
            AssertionError: If gitea_url ends with '/api/v1'.
        """
        assert not gitea_url.endswith('/api/v1')
        self.gitea_url = gitea_url + '/api/v1'
        self.session = requests.Session()
        self.session.headers['Content-Type'] = 'application/json'
        if token:
            self.session.headers['Authorization'] = f'token {token}'

    def get_default_branch_sha(self, owner: str, repo: str, branch: str) -> str:
        """Retrieve the commit SHA of the specified branch.

        Args:
            owner (str): Owner of the repository.
            repo (str): Name of the repository.
            branch (str): Name of the branch.

        Returns:
            str: The commit SHA of the specified branch.

        Raises:
            requests.HTTPError: If the API request fails.
        """
        url = f'{self.gitea_url}/repos/{owner}/{repo}/branches/{branch}'
        response = self.session.get(url)
        response.raise_for_status()
        data = response.json()
        return data['commit']['sha']

    def create_branch(self, owner: str, repo: str, new_branch: str, sha: str) -> bool:
        """Create a new branch from the provided SHA.

        Args:
            owner (str): Owner of the repository.
            repo (str): Name of the repository.
            new_branch (str): Name of the new branch to create.
            sha (str): Commit SHA to use as the starting point for the new branch.

        Returns:
            bool: True if the branch was created successfully, False if the branch already exists.

        Raises:
            requests.HTTPError: If the API request fails for reasons other than branch already existing.
        """
        url = f'{self.gitea_url}/repos/{owner}/{repo}/git/refs'
        json_data = {'ref': f'refs/heads/{new_branch}', 'sha': sha}
        response = self.session.post(url, json=json_data)
        if response.status_code == 422:
            logger.warning('Branch %s already exists.', new_branch)
            return False
        response.raise_for_status()
        return True

    def get_issues(self, owner: str, repo: str) -> list[dict[str, str]]:
        """Download issues from the specified repository and filter those with the 'aider' label.

        Args:
            owner (str): Owner of the repository.
            repo (str): Name of the repository.

        Returns:
            list: A list of issue dictionaries, filtered to only include issues with the 'aider' label.

        Raises:
            requests.HTTPError: If the API request fails.
        """
        url = f'{self.gitea_url}/repos/{owner}/{repo}/issues'
        response = self.session.get(url)
        response.raise_for_status()
        issues = response.json()
        # Filter to only include issues marked with the "aider" label.
        issues = [
            issue
            for issue in issues
            if any(label.get('name') == 'aider' for label in issue.get('labels', []))
        ]
        return issues

    def iter_user_repositories(
        self,
        owner: str,
        only_those_with_issues: bool = False,
    ) -> Iterator[str]:
        """Get a list of repositories for a given user.

        Args:
            owner (str): The owner of the repositories.
            only_those_with_issues (bool): If True, only return repositories with issues enabled.

        Returns:
            Iterator[str]: An iterator of repository names.
        """
        url = f'{self.gitea_url}/user/repos'
        response = self.session.get(url)
        response.raise_for_status()

        for repo in response.json():
            if only_those_with_issues and not repo['has_issues']:
                continue
            if repo['owner']['login'].lower() != owner.lower():
                continue
            yield repo['name']

    def create_pull_request(
        self,
        owner: str,
        repo: str,
        title: str,
        body: str,
        head: str,
        base: str,
        labels: list[str] = None,
    ) -> dict:
        """Create a pull request and optionally apply labels.

        Args:
            owner (str): Owner of the repository.
            repo (str): Name of the repository.
            title (str): Title of the pull request.
            body (str): Description/body of the pull request.
            head (str): The name of the branch where changes are implemented.
            base (str): The name of the branch you want the changes pulled into.
            labels (list[str], optional): List of label names to apply to the pull request.

        Returns:
            dict: The created pull request data.

        Raises:
            requests.HTTPError: If the API request fails.
        """
        url = f'{self.gitea_url}/repos/{owner}/{repo}/pulls'
        json_data = {
            'title': title,
            'body': body,
            'head': head,
            'base': base,
        }

        response = self.session.post(url, json=json_data)
        # If a pull request for this head/base already exists, return it instead of crashing
        if response.status_code == 422:
            logger.warning(
                'Pull request already exists for head %s and base %s', head, base,
            )
            prs = self.get_pull_requests(owner, repo)
            for pr in prs:
                if (
                    pr.get('head', {}).get('ref') == head
                    and pr.get('base', {}).get('ref') == base
                ):
                    return pr
            # fallback to raise if we can’t find it
            response.raise_for_status()
        response.raise_for_status()
        return response.json()

    def get_pull_request_comments(
        self,
        owner: str,
        repo: str,
        pr_number: str,
    ) -> list[dict]:
        """
        Fetch comments for a pull request.
        """
        url = f'{self.gitea_url}/repos/{owner}/{repo}/pulls/{pr_number}/comments'
        response = self.session.get(url)
        response.raise_for_status()
        return response.json()

    def get_pull_requests(
        self,
        owner: str,
        repo: str,
        state: str = 'open',
    ) -> list[dict]:
        """Fetch pull requests for a repository."""
        url = f'{self.gitea_url}/repos/{owner}/{repo}/pulls?state={state}'
        response = self.session.get(url)
        response.raise_for_status()
        return response.json()