tegwick
fd8f792f08
- Create new gitea/ package with clean API facade - Establish proper separation of concerns: tddai uses gitea, not vice versa - Replace duplicate curl+subprocess patterns with unified HTTP client - Add rich domain models with properties (issue.priority, issue.status) - Maintain full backwards compatibility in tddai modules - Reduce code complexity: -373 lines, +151 lines (net -222 lines) - Improve testability and maintainability through clean interfaces Architecture: - gitea.client.GiteaClient - main facade with sub-clients - gitea.api_client - high-level API with model conversion - gitea.http_client - low-level HTTP operations - gitea.models - rich domain objects (Issue, Milestone, Label) - gitea.config - gitea-specific configuration - gitea.exceptions - clean exception hierarchy 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
98 lines
3.5 KiB
Python
98 lines
3.5 KiB
Python
"""
|
|
Low-level HTTP client for Gitea API operations.
|
|
|
|
This module handles the actual HTTP requests to Gitea API using subprocess + curl
|
|
for maximum compatibility and minimal dependencies.
|
|
"""
|
|
|
|
import json
|
|
import subprocess
|
|
from subprocess import PIPE
|
|
from typing import Dict, Any, Optional, List
|
|
|
|
from .exceptions import GiteaError, GiteaApiError, GiteaAuthError
|
|
from .config import GiteaConfig
|
|
|
|
|
|
class GiteaHttpClient:
|
|
"""Low-level HTTP client for Gitea API."""
|
|
|
|
def __init__(self, config: GiteaConfig):
|
|
self.config = config
|
|
|
|
def get(self, url: str, params: Optional[Dict[str, str]] = None) -> Dict[str, Any]:
|
|
"""Make GET request to Gitea API."""
|
|
if params:
|
|
param_string = '&'.join(f"{k}={v}" for k, v in params.items())
|
|
url = f"{url}?{param_string}"
|
|
|
|
return self._make_request('GET', url)
|
|
|
|
def post(self, url: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
|
"""Make POST request to Gitea API."""
|
|
self._require_auth()
|
|
return self._make_request('POST', url, data)
|
|
|
|
def patch(self, url: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
|
"""Make PATCH request to Gitea API."""
|
|
self._require_auth()
|
|
return self._make_request('PATCH', url, data)
|
|
|
|
def delete(self, url: str) -> Dict[str, Any]:
|
|
"""Make DELETE request to Gitea API."""
|
|
self._require_auth()
|
|
return self._make_request('DELETE', url)
|
|
|
|
def _make_request(self, method: str, url: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
|
"""Make HTTP request using curl."""
|
|
cmd = ['curl', '-s', '-X', method]
|
|
|
|
# Add authentication if available
|
|
if self.config.auth_token:
|
|
cmd.extend(['-H', f'Authorization: token {self.config.auth_token}'])
|
|
|
|
# Add content type for requests with data
|
|
if data is not None:
|
|
cmd.extend(['-H', 'Content-Type: application/json'])
|
|
cmd.extend(['-d', json.dumps(data)])
|
|
|
|
cmd.append(url)
|
|
|
|
try:
|
|
result = subprocess.run(
|
|
cmd,
|
|
stdout=PIPE,
|
|
stderr=PIPE,
|
|
universal_newlines=True,
|
|
check=True
|
|
)
|
|
|
|
if result.returncode != 0:
|
|
raise GiteaApiError(f"HTTP request failed: {result.stderr}")
|
|
|
|
# Handle empty responses
|
|
if not result.stdout.strip():
|
|
return {}
|
|
|
|
response_data = json.loads(result.stdout)
|
|
|
|
# Check for API error responses
|
|
if isinstance(response_data, dict):
|
|
if 'message' in response_data:
|
|
# This could be an error or just a response with a message field
|
|
# We need to distinguish based on context or HTTP status
|
|
if any(error_word in response_data['message'].lower()
|
|
for error_word in ['error', 'not found', 'forbidden', 'unauthorized']):
|
|
raise GiteaApiError(response_data['message'])
|
|
|
|
return response_data
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
raise GiteaApiError(f"HTTP request failed: {e.stderr}")
|
|
except json.JSONDecodeError as e:
|
|
raise GiteaError(f"Failed to parse API response: {e}")
|
|
|
|
def _require_auth(self):
|
|
"""Ensure authentication token is available."""
|
|
if not self.config.auth_token:
|
|
raise GiteaAuthError("Authentication token required for this operation") |