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.
        """
        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.
        """
        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)
                }
                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.
        """
        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)
                }
        
        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.
        """
        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 {}
            
    def set_default_branch(self, owner: str, repo_name: str, branch_name: str):
        """
        Sets the default branch for a repository via the API.
        """
        self.logger.info(f"Setting default branch for '{owner}/{repo_name}' to '{branch_name}'...")
        endpoint = f"/repos/{owner}/{repo_name}"
        payload = {
            "default_branch": branch_name
        }
        
        try:
            self._api_request("PATCH", endpoint, json_data=payload)
            self.logger.info("Default branch set successfully.")
        except GiteaAPIError as e:
            self.logger.error(f"Failed to set default branch for '{repo_name}': {e}")
            # This is not a critical failure, so we log it but don't re-raise.