82 lines
3 KiB
Python
82 lines
3 KiB
Python
|
"""
|
||
|
|
Issue writing to Gitea API.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import subprocess
|
||
|
|
from subprocess import PIPE
|
||
|
|
from typing import Dict, Any, Optional
|
||
|
|
|
||
|
|
from .config import get_config
|
||
|
|
from .exceptions import IssueError
|
||
|
|
|
||
|
|
|
||
|
|
class IssueWriter:
|
||
|
|
"""Writes issue updates to Gitea API."""
|
||
|
|
|
||
|
|
def __init__(self, config=None, auth_token=None):
|
||
|
|
self.config = config or get_config()
|
||
|
|
self.auth_token = auth_token or os.getenv('GITEA_TOKEN')
|
||
|
|
|
||
|
|
def update_issue(self, issue_number: int, update_data: Dict[str, Any]) -> Dict[str, Any]:
|
||
|
|
"""Update an issue via PATCH operation."""
|
||
|
|
if not self.auth_token:
|
||
|
|
raise IssueError("Authentication token required for issue updates")
|
||
|
|
|
||
|
|
url = f"{self.config.issues_api_url}/{issue_number}"
|
||
|
|
|
||
|
|
try:
|
||
|
|
# Prepare curl command with authentication
|
||
|
|
curl_cmd = [
|
||
|
|
'curl', '-s', '-X', 'PATCH',
|
||
|
|
'-H', 'Content-Type: application/json',
|
||
|
|
'-H', f'Authorization: token {self.auth_token}',
|
||
|
|
'-d', json.dumps(update_data),
|
||
|
|
url
|
||
|
|
]
|
||
|
|
|
||
|
|
result = subprocess.run(
|
||
|
|
curl_cmd,
|
||
|
|
stdout=PIPE,
|
||
|
|
stderr=PIPE,
|
||
|
|
universal_newlines=True,
|
||
|
|
check=True
|
||
|
|
)
|
||
|
|
|
||
|
|
if result.returncode != 0:
|
||
|
|
raise IssueError(f"Failed to update issue #{issue_number}: {result.stderr}")
|
||
|
|
|
||
|
|
response_data = json.loads(result.stdout)
|
||
|
|
|
||
|
|
if 'message' in response_data and 'number' not in response_data:
|
||
|
|
raise IssueError(f"Failed to update issue #{issue_number}: {response_data['message']}")
|
||
|
|
|
||
|
|
return response_data
|
||
|
|
|
||
|
|
except subprocess.CalledProcessError as e:
|
||
|
|
raise IssueError(f"Failed to update issue #{issue_number}: {e}")
|
||
|
|
except json.JSONDecodeError as e:
|
||
|
|
raise IssueError(f"Failed to parse response data: {e}")
|
||
|
|
|
||
|
|
def update_issue_title(self, issue_number: int, new_title: str) -> Dict[str, Any]:
|
||
|
|
"""Update only the title of an issue."""
|
||
|
|
return self.update_issue(issue_number, {'title': new_title})
|
||
|
|
|
||
|
|
def update_issue_body(self, issue_number: int, new_body: str) -> Dict[str, Any]:
|
||
|
|
"""Update only the body of an issue."""
|
||
|
|
return self.update_issue(issue_number, {'body': new_body})
|
||
|
|
|
||
|
|
def update_issue_state(self, issue_number: int, new_state: str) -> Dict[str, Any]:
|
||
|
|
"""Update only the state of an issue (open/closed)."""
|
||
|
|
if new_state not in ['open', 'closed']:
|
||
|
|
raise IssueError(f"Invalid state '{new_state}'. Must be 'open' or 'closed'")
|
||
|
|
return self.update_issue(issue_number, {'state': new_state})
|
||
|
|
|
||
|
|
def close_issue(self, issue_number: int) -> Dict[str, Any]:
|
||
|
|
"""Close an issue."""
|
||
|
|
return self.update_issue_state(issue_number, 'closed')
|
||
|
|
|
||
|
|
def reopen_issue(self, issue_number: int) -> Dict[str, Any]:
|
||
|
|
"""Reopen a closed issue."""
|
||
|
|
return self.update_issue_state(issue_number, 'open')
|