SXXXXXXX_RepoSync/reposync/core/gitea_client.py

# RepoSync/core/gitea_client.py

"""
Gitea-specific implementation of the BaseVCSClient.

This module handles all interactions with a Gitea instance via its REST API.
"""

import logging
import requests
from typing import List, Dict, Optional, Any

from .base_vcs_client import BaseVCSClient


class GiteaAPIError(Exception):
    """Custom exception for Gitea API errors."""

    pass


class GiteaClient(BaseVCSClient):
    """
    A client for interacting with the Gitea API.
    Implements the abstract methods defined in BaseVCSClient.
    """

    def __init__(self, api_url: str, token: str, logger: logging.Logger):
        """
        Initializes the GiteaClient.

        Args:
            api_url: The base URL of the Gitea instance (e.g., 'https://gitea.com').
            token: The personal access token for authentication.
            logger: A configured logger instance.
        """
        super().__init__(api_url, token)
        self.logger = logger
        self.logger.debug(f"GiteaClient initialized for URL: {self.api_url}")

    def _api_request(
        self,
        method: str,
        endpoint: str,
        params: Optional[Dict[str, Any]] = None,
        json_data: Optional[Dict[str, Any]] = None,
    ) -> Any:
        """
        A helper method to make requests to the Gitea API.
        """
        url = f"{self.api_url}/api/v1{endpoint}"
        self.logger.debug(f"Making API call: {method} {url}")

        try:
            response = requests.request(
                method=method,
                url=url,
                headers=self.headers,
                params=params,
                json=json_data,
                timeout=30,
            )
            response.raise_for_status()

            if response.status_code == 204:
                return None

            return response.json()
        except requests.exceptions.RequestException as e:
            self.logger.error(f"Gitea API request failed: {e}")
            raise GiteaAPIError(f"Failed to communicate with Gitea API: {e}") from e

    def get_repositories(self) -> List[Dict[str, Any]]:
        """
        Retrieves a list of all repositories accessible by the user, handling pagination.
        Now includes repository size.
        """
        self.logger.info("Fetching list of repositories from Gitea...")
        all_repos = []
        page = 1
        limit = 50

        while True:
            self.logger.debug(f"Fetching page {page} of repositories...")
            params = {"page": page, "limit": limit}
            repos_page = self._api_request("GET", "/user/repos", params=params)

            if not repos_page:
                break

            for repo in repos_page:
                standardized_repo = {
                    "name": repo.get("name"),
                    "owner": repo.get("owner", {}).get("login"),
                    "clone_url": repo.get("clone_url"),
                    "description": repo.get("description"),
                    "private": repo.get("private"),
                    "size_kb": repo.get("size", 0),  # Get size in KB
                }
                all_repos.append(standardized_repo)

            page += 1

        self.logger.info(f"Found {len(all_repos)} repositories.")
        return all_repos

    def get_repository(self, repo_name: str) -> Optional[Dict[str, Any]]:
        """
        Retrieves a single repository by its name using the search endpoint.
        Now includes repository size.
        """
        self.logger.info(f"Searching for repository '{repo_name}'...")
        params = {"q": repo_name, "limit": 10}

        try:
            search_result = self._api_request("GET", "/repos/search", params=params)
        except GiteaAPIError as e:
            self.logger.warning(
                f"Repo search failed: {e}. This may be a permissions issue."
            )
            return None

        if not search_result or not search_result.get("data"):
            self.logger.info(f"Repository '{repo_name}' not found via search.")
            return None

        for repo in search_result["data"]:
            if repo.get("name").lower() == repo_name.lower():
                self.logger.info(f"Found exact match for repository '{repo_name}'.")
                return {
                    "name": repo.get("name"),
                    "owner": repo.get("owner", {}).get("login"),
                    "clone_url": repo.get("clone_url"),
                    "description": repo.get("description"),
                    "private": repo.get("private"),
                    "size_kb": repo.get("size", 0),  # Get size in KB
                }

        self.logger.info(
            f"No exact match for repository '{repo_name}' found in search results."
        )
        return None

    def create_repository(
        self, name: str, description: str = "", private: bool = True
    ) -> Dict[str, Any]:
        """
        Creates a new repository on the Gitea platform.
        """
        self.logger.info(f"Creating new repository '{name}'...")
        payload = {"name": name, "description": description, "private": private}
        new_repo = self._api_request("POST", "/user/repos", json_data=payload)

        self.logger.info(f"Successfully created repository '{name}'.")
        return {
            "name": new_repo.get("name"),
            "owner": new_repo.get("owner", {}).get("login"),
            "clone_url": new_repo.get("clone_url"),
            "description": new_repo.get("description"),
            "private": new_repo.get("private"),
            "size_kb": new_repo.get("size", 0),
        }

    def get_repository_branches(self, owner: str, repo_name: str) -> Dict[str, str]:
        """
        Retrieves a dictionary of branches and their head commit hashes for a
        specific repository on Gitea.
        """
        self.logger.info(f"Fetching branches for '{owner}/{repo_name}' from Gitea...")
        endpoint = f"/repos/{owner}/{repo_name}/branches"

        try:
            branches_data = self._api_request("GET", endpoint)
            branches = {}
            if branches_data:
                for branch in branches_data:
                    branch_name = branch.get("name")
                    commit_hash = branch.get("commit", {}).get("id")
                    if branch_name and commit_hash:
                        branches[branch_name] = commit_hash

            self.logger.debug(
                f"Found branches for '{repo_name}': {list(branches.keys())}"
            )
            return branches
        except GiteaAPIError as e:
            self.logger.error(f"Failed to fetch branches for '{repo_name}': {e}")
            return {}