131 lines
7.5 KiB
Python
131 lines
7.5 KiB
Python
# RepoSync/core/gitea_client.py
|
|
|
|
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):
|
|
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, raise_on_404: bool = True) -> Any:
|
|
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)
|
|
if not raise_on_404 and response.status_code == 404: return None
|
|
response.raise_for_status()
|
|
return None if response.status_code == 204 else 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]]:
|
|
self.logger.info("Fetching list of repositories and wikis 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:
|
|
all_repos.append({
|
|
"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), "is_wiki": False
|
|
})
|
|
if repo.get("has_wiki"):
|
|
repo_name, clone_url = repo.get("name"), repo.get("clone_url")
|
|
if clone_url and clone_url.endswith(".git"):
|
|
wiki_clone_url = clone_url[:-4] + ".wiki.git"
|
|
all_repos.append({
|
|
"name": f"{repo_name} (Wiki)", "owner": repo.get("owner", {}).get("login"),
|
|
"clone_url": wiki_clone_url, "description": f"Wiki for the '{repo_name}' repository",
|
|
"private": repo.get("private"), "size_kb": 0, "is_wiki": True
|
|
})
|
|
page += 1
|
|
self.logger.info(f"Found {len(all_repos)} items (repositories and wikis).")
|
|
return all_repos
|
|
|
|
def get_repository(self, repo_name: str) -> Optional[Dict[str, Any]]:
|
|
self.logger.info(f"Searching for item '{repo_name}'...")
|
|
is_wiki, base_repo_name = repo_name.endswith(" (Wiki)"), repo_name.removesuffix(" (Wiki)").strip()
|
|
params = {"q": base_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}."); return None
|
|
if not search_result or not search_result.get("data"):
|
|
self.logger.info(f"Item '{base_repo_name}' not found via search."); return None
|
|
for repo in search_result["data"]:
|
|
if repo.get("name").lower() == base_repo_name.lower():
|
|
self.logger.info(f"Found base repository '{base_repo_name}'.")
|
|
owner, clone_url = repo.get("owner", {}).get("login"), repo.get("clone_url")
|
|
if is_wiki:
|
|
wiki_clone_url = f"{self.api_url}/{owner}/{base_repo_name}.wiki.git"
|
|
return {"name": repo_name, "owner": owner, "clone_url": wiki_clone_url, "is_wiki": True}
|
|
else:
|
|
return {"name": repo.get("name"), "owner": owner, "clone_url": clone_url, "description": repo.get("description"), "private": repo.get("private"), "size_kb": repo.get("size", 0), "is_wiki": False}
|
|
self.logger.info(f"No exact match for '{base_repo_name}' found in search results."); return None
|
|
|
|
def create_repository(self, name: str, description: str = "", private: bool = True, is_wiki: bool = False) -> Dict[str, Any]:
|
|
self.logger.info(f"Request to create item: '{name}' (is_wiki={is_wiki})")
|
|
if is_wiki:
|
|
# Creation of wiki repo is handled by the git push process in SyncManager.
|
|
# This function should only be called for non-wiki repos in the import flow.
|
|
raise GiteaAPIError("Wiki creation should be handled by the sync manager's Git-based initialization.")
|
|
|
|
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), "is_wiki": False}
|
|
|
|
def get_repository_branches(self, owner: str, repo_name: str) -> Dict[str, str]:
|
|
self.logger.info(f"Fetching branches for '{owner}/{repo_name}' from Gitea...")
|
|
api_repo_name = repo_name.removesuffix(" (Wiki)").strip()
|
|
if repo_name.endswith(" (Wiki)"): api_repo_name += ".wiki"
|
|
endpoint = f"/repos/{owner}/{api_repo_name}/branches"
|
|
try:
|
|
branches_data = self._api_request("GET", endpoint, raise_on_404=False)
|
|
if branches_data is None: return {}
|
|
return {b.get("name"): b.get("commit", {}).get("id") for b in branches_data if b.get("name") and b.get("commit", {}).get("id")}
|
|
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):
|
|
api_repo_name = repo_name.removesuffix(" (Wiki)").strip()
|
|
endpoint = f"/repos/{owner}/{api_repo_name}"
|
|
payload = {"default_branch": branch_name}
|
|
if repo_name.endswith(" (Wiki)"):
|
|
payload = {"wiki": {"default_branch": branch_name}}
|
|
|
|
try:
|
|
self._api_request("PATCH", endpoint, json_data=payload)
|
|
self.logger.info(f"Default branch for '{repo_name}' set to '{branch_name}'.")
|
|
except GiteaAPIError as e: self.logger.error(f"Failed to set default branch for '{repo_name}': {e}")
|
|
|
|
def enable_wiki(self, owner: str, repo_name: str):
|
|
"""Ensures the wiki feature is enabled for a repository via API."""
|
|
self.logger.info(f"Ensuring wiki feature is enabled for '{owner}/{repo_name}'...")
|
|
endpoint_repo_edit = f"/repos/{owner}/{repo_name}"
|
|
payload_enable = {"has_wiki": True}
|
|
try:
|
|
self._api_request("PATCH", endpoint_repo_edit, json_data=payload_enable)
|
|
self.logger.info(f"Wiki feature successfully enabled for '{owner}/{repo_name}'.")
|
|
except GiteaAPIError as e:
|
|
self.logger.error(f"Failed to enable wiki flag for '{repo_name}': {e}")
|
|
raise |