Bernd Worsch
7b9d6af5a1
- Replace walrus operator (:=) with traditional assignment in config.py - Replace datetime.fromisoformat() with strptime() for Python 3.6 - Replace subprocess capture_output and text params with PIPE and universal_newlines - All tests now pass on Python 3.6.9 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
139 lines
4.7 KiB
Python
139 lines
4.7 KiB
Python
"""
|
|
Issue fetching from Gitea API.
|
|
"""
|
|
|
|
import json
|
|
import subprocess
|
|
from subprocess import PIPE
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from typing import List, Optional, Dict, Any
|
|
|
|
from .config import get_config
|
|
from .exceptions import IssueError
|
|
|
|
|
|
@dataclass
|
|
class Issue:
|
|
"""Represents a Gitea issue."""
|
|
|
|
number: int
|
|
title: str
|
|
body: str
|
|
state: str
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
html_url: str
|
|
assignee: Optional[str] = None
|
|
labels: List[str] = None
|
|
|
|
def __post_init__(self):
|
|
if self.labels is None:
|
|
self.labels = []
|
|
|
|
|
|
class IssueFetcher:
|
|
"""Fetches issues from Gitea API."""
|
|
|
|
def __init__(self, config=None):
|
|
self.config = config or get_config()
|
|
|
|
def fetch_issue(self, issue_number: int) -> Issue:
|
|
"""Fetch a specific issue by number."""
|
|
try:
|
|
result = subprocess.run(
|
|
['curl', '-s', f"{self.config.issues_api_url}/{issue_number}"],
|
|
stdout=PIPE,
|
|
stderr=PIPE,
|
|
universal_newlines=True,
|
|
check=True
|
|
)
|
|
|
|
if result.returncode != 0:
|
|
raise IssueError(f"Failed to fetch issue #{issue_number}: {result.stderr}")
|
|
|
|
issue_data = json.loads(result.stdout)
|
|
|
|
if 'message' in issue_data:
|
|
raise IssueError(f"Issue #{issue_number} not found: {issue_data['message']}")
|
|
|
|
return self._parse_issue(issue_data)
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
raise IssueError(f"Failed to fetch issue #{issue_number}: {e}")
|
|
except json.JSONDecodeError as e:
|
|
raise IssueError(f"Failed to parse issue data: {e}")
|
|
|
|
def fetch_issues(self, state: str = "all") -> List[Issue]:
|
|
"""Fetch all issues with optional state filter."""
|
|
try:
|
|
url = self.config.issues_api_url
|
|
if state != "all":
|
|
url += f"?state={state}"
|
|
|
|
result = subprocess.run(
|
|
['curl', '-s', url],
|
|
stdout=PIPE,
|
|
stderr=PIPE,
|
|
universal_newlines=True,
|
|
check=True
|
|
)
|
|
|
|
if result.returncode != 0:
|
|
raise IssueError(f"Failed to fetch issues: {result.stderr}")
|
|
|
|
issues_data = json.loads(result.stdout)
|
|
|
|
if isinstance(issues_data, dict) and 'message' in issues_data:
|
|
raise IssueError(f"Failed to fetch issues: {issues_data['message']}")
|
|
|
|
if not isinstance(issues_data, list):
|
|
raise IssueError("Invalid response format: expected list of issues")
|
|
|
|
return [self._parse_issue(issue_data) for issue_data in issues_data]
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
raise IssueError(f"Failed to fetch issues: {e}")
|
|
except json.JSONDecodeError as e:
|
|
raise IssueError(f"Failed to parse issues data: {e}")
|
|
|
|
def fetch_open_issues(self) -> List[Issue]:
|
|
"""Fetch only open issues."""
|
|
return self.fetch_issues(state="open")
|
|
|
|
def _parse_issue(self, issue_data: Dict[str, Any]) -> Issue:
|
|
"""Parse issue data from API response."""
|
|
try:
|
|
labels = [label['name'] for label in issue_data.get('labels', [])]
|
|
assignee = None
|
|
if issue_data.get('assignee'):
|
|
assignee = issue_data['assignee'].get('login')
|
|
|
|
return Issue(
|
|
number=issue_data['number'],
|
|
title=issue_data['title'],
|
|
body=issue_data.get('body', ''),
|
|
state=issue_data['state'],
|
|
created_at=datetime.strptime(issue_data['created_at'].replace('Z', '').split('.')[0], '%Y-%m-%dT%H:%M:%S'),
|
|
updated_at=datetime.strptime(issue_data['updated_at'].replace('Z', '').split('.')[0], '%Y-%m-%dT%H:%M:%S'),
|
|
html_url=issue_data['html_url'],
|
|
assignee=assignee,
|
|
labels=labels
|
|
)
|
|
except (KeyError, ValueError) as e:
|
|
raise IssueError(f"Failed to parse issue data: {e}")
|
|
|
|
def get_issue_data_dict(self, issue_number: int) -> Dict[str, Any]:
|
|
"""Get issue data as dictionary for workspace creation."""
|
|
issue = self.fetch_issue(issue_number)
|
|
return {
|
|
'number': issue.number,
|
|
'title': issue.title,
|
|
'body': issue.body,
|
|
'state': issue.state,
|
|
'created_at': issue.created_at.isoformat(),
|
|
'updated_at': issue.updated_at.isoformat(),
|
|
'html_url': issue.html_url,
|
|
'assignee': {'login': issue.assignee} if issue.assignee else None,
|
|
'labels': [{'name': label} for label in issue.labels]
|
|
} |