tegwick
d0ffdc057c
- Move release management to capabilities/release-management/ with complete Makefile - Create automatic capability discovery system in scripts/capability_discovery.mk - Add capability-manager subagent for managing modular architecture - Implement target delegation system enabling capability-name-target patterns - Create Makefiles for markitect-content, markitect-utils, and issue-facade capabilities - Remove legacy release management code and documentation from main project - Update main Makefile to use capability discovery and delegation - Add comprehensive capability status, help, and management targets The capability system provides: - Automatic discovery of capabilities with Makefiles - Clean target delegation without conflicts - Modular architecture following established patterns - Comprehensive help and status reporting - Zero-conflict capability integration 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
355 lines
13 KiB
Python
355 lines
13 KiB
Python
"""
|
|
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
|
|
from pathlib import Path
|
|
from typing import Optional, List, Dict, Any
|
|
|
|
from ..base import RegistryInterface
|
|
from .config import GiteaConfig
|
|
from .exceptions import GiteaError
|
|
|
|
|
|
class GiteaRegistry(RegistryInterface):
|
|
"""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[str, Any]]:
|
|
"""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[str, Any]]:
|
|
"""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, package_path: Path, dry_run: bool = False) -> bool:
|
|
"""Upload a package to Gitea registry.
|
|
|
|
Args:
|
|
package_path: Path to package file (.whl or .tar.gz)
|
|
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 package_path.exists():
|
|
raise GiteaError(f"Package file not found: {package_path}")
|
|
|
|
if dry_run:
|
|
print(f"[DRY RUN] Would upload to: {self.pypi_registry_url}")
|
|
print(f"[DRY RUN] Would upload: {package_path}")
|
|
return True
|
|
|
|
return self._upload_file(package_path)
|
|
|
|
def upload_package_as_release_assets(self,
|
|
version: str,
|
|
wheel_path: Path,
|
|
sdist_path: Optional[Path] = None,
|
|
dry_run: bool = False) -> bool:
|
|
"""Upload packages as Gitea release assets (fallback when package registry unavailable).
|
|
|
|
Args:
|
|
version: Version tag (e.g., "v0.8.0")
|
|
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 release 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 release assets for {version}")
|
|
print(f"[DRY RUN] Release API: {self.config.repo_api_url}/releases")
|
|
for file_path in files_to_upload:
|
|
print(f"[DRY RUN] Would upload: {file_path}")
|
|
return True
|
|
|
|
# Create or get release
|
|
release_id = self._create_or_get_release(version)
|
|
if not release_id:
|
|
return False
|
|
|
|
# Upload each file as release asset
|
|
success = True
|
|
for file_path in files_to_upload:
|
|
if not self._upload_release_asset(release_id, file_path):
|
|
success = False
|
|
|
|
return success
|
|
|
|
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[str, Any]:
|
|
"""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 _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
|
|
|
|
# Gitea PyPI upload API expects PUT with the file content as body
|
|
# URL format: /api/packages/{owner}/pypi/{filename}
|
|
upload_url = f"{self.config.gitea_url}/api/packages/{self.config.repo_owner}/pypi"
|
|
|
|
with open(file_path, 'rb') as f:
|
|
file_content = f.read()
|
|
|
|
headers = {
|
|
'Authorization': f'token {self.config.auth_token}',
|
|
'Content-Type': 'application/octet-stream'
|
|
}
|
|
|
|
# Upload using PUT request with filename in URL
|
|
upload_endpoint = f"{upload_url}/{file_path.name}"
|
|
response = requests.put(
|
|
upload_endpoint,
|
|
headers=headers,
|
|
data=file_content,
|
|
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}")
|
|
if response.text:
|
|
print(f" Error: {response.text}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
print(f"❌ Upload failed for {file_path.name}: {e}")
|
|
return False
|
|
|
|
def _create_or_get_release(self, version: str) -> Optional[int]:
|
|
"""Create a new release or get existing release ID.
|
|
|
|
Args:
|
|
version: Version tag (e.g., "v0.8.0")
|
|
|
|
Returns:
|
|
Release ID if successful, None otherwise
|
|
"""
|
|
try:
|
|
import requests
|
|
|
|
# Ensure version has 'v' prefix
|
|
tag_name = version if version.startswith('v') else f'v{version}'
|
|
|
|
headers = {"Authorization": f"token {self.config.auth_token}"}
|
|
|
|
# First, try to get existing release
|
|
releases_url = f"{self.config.repo_api_url}/releases"
|
|
response = requests.get(releases_url, headers=headers, timeout=10)
|
|
|
|
if response.status_code == 200:
|
|
releases = response.json()
|
|
for release in releases:
|
|
if release.get('tag_name') == tag_name:
|
|
print(f"✅ Found existing release: {tag_name}")
|
|
return release['id']
|
|
|
|
# Create new release
|
|
release_data = {
|
|
"tag_name": tag_name,
|
|
"name": f"MarkiTect {version.lstrip('v')}",
|
|
"body": f"Release {version.lstrip('v')}\\n\\nPython packages for MarkiTect.",
|
|
"draft": False,
|
|
"prerelease": False
|
|
}
|
|
|
|
response = requests.post(releases_url, headers=headers, json=release_data, timeout=10)
|
|
|
|
if response.status_code == 201:
|
|
release = response.json()
|
|
print(f"✅ Created release: {tag_name}")
|
|
return release['id']
|
|
else:
|
|
print(f"❌ Failed to create release: {response.status_code} {response.text}")
|
|
return None
|
|
|
|
except Exception as e:
|
|
print(f"❌ Error managing release: {e}")
|
|
return None
|
|
|
|
def _upload_release_asset(self, release_id: int, file_path: Path) -> bool:
|
|
"""Upload a file as a release asset.
|
|
|
|
Args:
|
|
release_id: Gitea release ID
|
|
file_path: Path to file to upload
|
|
|
|
Returns:
|
|
True if upload successful, False otherwise
|
|
"""
|
|
try:
|
|
import requests
|
|
|
|
# Upload asset to Gitea release
|
|
upload_url = f"{self.config.repo_api_url}/releases/{release_id}/assets"
|
|
|
|
headers = {
|
|
"Authorization": f"token {self.config.auth_token}"
|
|
}
|
|
|
|
with open(file_path, 'rb') as f:
|
|
files = {
|
|
'attachment': (file_path.name, f, 'application/octet-stream')
|
|
}
|
|
|
|
response = requests.post(
|
|
upload_url,
|
|
headers=headers,
|
|
files=files,
|
|
timeout=120 # Larger timeout for file uploads
|
|
)
|
|
|
|
if response.status_code == 201:
|
|
print(f"✅ Uploaded release asset: {file_path.name}")
|
|
return True
|
|
else:
|
|
print(f"❌ Failed to upload {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 |