""" Gitea Package Registry Client This module provides functionality to publish Python packages to Gitea's package registry. Gitea supports multiple package registries including PyPI-compatible registries. """ import os import subprocess import tempfile from pathlib import Path from typing import Optional, List, Dict from .config import GiteaConfig from .exceptions import GiteaError class GiteaPackageRegistry: """Client for publishing packages to Gitea package registry.""" def __init__(self, config: Optional[GiteaConfig] = None): """Initialize the package registry client. Args: config: Gitea configuration. If None, auto-detects from git repository. """ self.config = config or GiteaConfig.from_git_repository() self.config.validate() @property def pypi_registry_url(self) -> str: """Get the PyPI-compatible registry URL for this repository.""" return f"{self.config.gitea_url}/api/packages/{self.config.repo_owner}/pypi" @property def package_list_url(self) -> str: """Get the package listing URL for this repository.""" return f"{self.config.gitea_url}/api/v1/packages/{self.config.repo_owner}" def check_auth(self) -> bool: """Check if authentication token is available and valid.""" if not self.config.auth_token: return False try: # Test auth by trying to access packages API import requests headers = {"Authorization": f"token {self.config.auth_token}"} response = requests.get(self.package_list_url, headers=headers, timeout=10) return response.status_code in [200, 404] # 404 is okay if no packages exist yet except Exception: return False def list_packages(self) -> List[Dict]: """List all packages for this repository owner. Returns: List of package information dictionaries """ try: import requests headers = {} if self.config.auth_token: headers["Authorization"] = f"token {self.config.auth_token}" response = requests.get(self.package_list_url, headers=headers, timeout=10) response.raise_for_status() return response.json() except Exception as e: raise GiteaError(f"Failed to list packages: {e}") def get_package_info(self, package_name: str) -> Optional[Dict]: """Get information about a specific package. Args: package_name: Name of the package Returns: Package information dictionary or None if not found """ try: packages = self.list_packages() for package in packages: if package.get("name") == package_name: return package return None except Exception: return None def upload_package(self, wheel_path: Path, sdist_path: Optional[Path] = None, dry_run: bool = False) -> bool: """Upload package files to Gitea registry. Args: wheel_path: Path to wheel (.whl) file sdist_path: Optional path to source distribution (.tar.gz) file dry_run: If True, show what would be done without uploading Returns: True if upload successful, False otherwise """ if not self.config.auth_token: raise GiteaError("Authentication token required for package upload. Set GITEA_API_TOKEN environment variable.") if not wheel_path.exists(): raise GiteaError(f"Wheel file not found: {wheel_path}") if sdist_path and not sdist_path.exists(): raise GiteaError(f"Source distribution file not found: {sdist_path}") files_to_upload = [wheel_path] if sdist_path: files_to_upload.append(sdist_path) if dry_run: print(f"[DRY RUN] Would upload to: {self.pypi_registry_url}") for file_path in files_to_upload: print(f"[DRY RUN] Would upload: {file_path}") return True success = True for file_path in files_to_upload: if not self._upload_file(file_path): success = False return success def _upload_file(self, file_path: Path) -> bool: """Upload a single file to the registry. Args: file_path: Path to file to upload Returns: True if upload successful, False otherwise """ try: import requests # Upload using multipart form data (PyPI-compatible) with open(file_path, 'rb') as f: files = { 'content': (file_path.name, f, 'application/octet-stream') } headers = { 'Authorization': f'token {self.config.auth_token}' } upload_url = f"{self.pypi_registry_url}/simple/" response = requests.post( upload_url, headers=headers, files=files, timeout=60 ) if response.status_code in [200, 201, 409]: # 409 = already exists print(f"✅ Uploaded: {file_path.name}") if response.status_code == 409: print(f" (already exists)") return True else: print(f"❌ Upload failed for {file_path.name}: {response.status_code} {response.text}") return False except Exception as e: print(f"❌ Upload failed for {file_path.name}: {e}") return False def delete_package_version(self, package_name: str, version: str, dry_run: bool = False) -> bool: """Delete a specific version of a package. Args: package_name: Name of the package version: Version to delete dry_run: If True, show what would be done without deleting Returns: True if deletion successful, False otherwise """ if not self.config.auth_token: raise GiteaError("Authentication token required for package deletion.") delete_url = f"{self.config.gitea_url}/api/v1/packages/{self.config.repo_owner}/pypi/{package_name}/{version}" if dry_run: print(f"[DRY RUN] Would delete: {package_name} v{version}") print(f"[DRY RUN] DELETE {delete_url}") return True try: import requests headers = {"Authorization": f"token {self.config.auth_token}"} response = requests.delete(delete_url, headers=headers, timeout=10) if response.status_code in [200, 204, 404]: # 404 = already deleted print(f"✅ Deleted: {package_name} v{version}") return True else: print(f"❌ Delete failed: {response.status_code} {response.text}") return False except Exception as e: print(f"❌ Delete failed: {e}") return False def get_registry_info(self) -> Dict: """Get information about the package registry configuration. Returns: Dictionary with registry information """ return { "gitea_url": self.config.gitea_url, "repo_owner": self.config.repo_owner, "repo_name": self.config.repo_name, "pypi_registry_url": self.pypi_registry_url, "package_list_url": self.package_list_url, "auth_configured": bool(self.config.auth_token), "auth_valid": self.check_auth() if self.config.auth_token else False } def configure_pip_for_gitea(config: Optional[GiteaConfig] = None, pip_conf_path: Optional[Path] = None) -> Path: """Configure pip to use Gitea package registry as additional index. Args: config: Gitea configuration pip_conf_path: Custom path for pip.conf file Returns: Path to created/updated pip.conf file """ config = config or GiteaConfig.from_git_repository() if pip_conf_path is None: # Default pip config location pip_conf_path = Path.home() / ".pip" / "pip.conf" pip_conf_path.parent.mkdir(parents=True, exist_ok=True) registry = GiteaPackageRegistry(config) gitea_index = f"{registry.pypi_registry_url}/simple/" # Read existing config or create new config_content = "" if pip_conf_path.exists(): config_content = pip_conf_path.read_text() # Add Gitea index if not already present if "extra-index-url" not in config_content: if "[global]" not in config_content: config_content = "[global]\n" + config_content lines = config_content.split('\n') global_section_idx = next(i for i, line in enumerate(lines) if line.strip() == "[global]") lines.insert(global_section_idx + 1, f"extra-index-url = {gitea_index}") config_content = '\n'.join(lines) elif gitea_index not in config_content: # Add to existing extra-index-url lines = config_content.split('\n') for i, line in enumerate(lines): if line.startswith("extra-index-url"): lines[i] = f"{line} {gitea_index}" break config_content = '\n'.join(lines) pip_conf_path.write_text(config_content) return pip_conf_path