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>
203 lines
7.3 KiB
Python
203 lines
7.3 KiB
Python
"""
|
|
High-level API client that converts between API responses and domain models.
|
|
"""
|
|
|
|
from datetime import datetime
|
|
from typing import List, Optional, Dict, Any
|
|
|
|
from .http_client import GiteaHttpClient
|
|
from .models import Issue, Milestone, Label, User, IssueCreateData, IssueUpdateData, MilestoneCreateData, LabelCreateData
|
|
from .config import GiteaConfig
|
|
from .exceptions import GiteaNotFoundError, GiteaError
|
|
|
|
|
|
class GiteaApiClient:
|
|
"""High-level API client with domain model conversion."""
|
|
|
|
def __init__(self, config: GiteaConfig):
|
|
self.config = config
|
|
self.http = GiteaHttpClient(config)
|
|
|
|
# Issue operations
|
|
def get_issue(self, issue_number: int) -> Issue:
|
|
"""Get a specific issue by number."""
|
|
try:
|
|
url = f"{self.config.issues_api_url}/{issue_number}"
|
|
data = self.http.get(url)
|
|
return self._parse_issue(data)
|
|
except GiteaError as e:
|
|
if "not found" in str(e).lower():
|
|
raise GiteaNotFoundError(f"Issue #{issue_number} not found")
|
|
raise
|
|
|
|
def list_issues(self, state: str = "all", page: int = 1, per_page: int = 50) -> List[Issue]:
|
|
"""List issues with optional filtering."""
|
|
params = {"page": str(page), "limit": str(per_page)}
|
|
if state != "all":
|
|
params["state"] = state
|
|
|
|
data = self.http.get(self.config.issues_api_url, params)
|
|
|
|
if not isinstance(data, list):
|
|
raise GiteaError("Invalid response format: expected list of issues")
|
|
|
|
return [self._parse_issue(issue_data) for issue_data in data]
|
|
|
|
def create_issue(self, issue_data: IssueCreateData) -> Issue:
|
|
"""Create a new issue."""
|
|
payload = {
|
|
"title": issue_data.title,
|
|
"body": issue_data.body,
|
|
}
|
|
|
|
if issue_data.assignees:
|
|
payload["assignees"] = issue_data.assignees
|
|
if issue_data.milestone:
|
|
payload["milestone"] = issue_data.milestone
|
|
if issue_data.labels:
|
|
payload["labels"] = issue_data.labels
|
|
|
|
data = self.http.post(self.config.issues_api_url, payload)
|
|
return self._parse_issue(data)
|
|
|
|
def update_issue(self, issue_number: int, update_data: IssueUpdateData) -> Issue:
|
|
"""Update an existing issue."""
|
|
payload = {}
|
|
|
|
if update_data.title is not None:
|
|
payload["title"] = update_data.title
|
|
if update_data.body is not None:
|
|
payload["body"] = update_data.body
|
|
if update_data.state is not None:
|
|
payload["state"] = update_data.state
|
|
if update_data.assignees is not None:
|
|
payload["assignees"] = update_data.assignees
|
|
if update_data.milestone is not None:
|
|
payload["milestone"] = update_data.milestone
|
|
if update_data.labels is not None:
|
|
payload["labels"] = update_data.labels
|
|
|
|
url = f"{self.config.issues_api_url}/{issue_number}"
|
|
data = self.http.patch(url, payload)
|
|
return self._parse_issue(data)
|
|
|
|
# Milestone operations
|
|
def list_milestones(self, state: str = "all") -> List[Milestone]:
|
|
"""List repository milestones."""
|
|
params = {}
|
|
if state != "all":
|
|
params["state"] = state
|
|
|
|
data = self.http.get(self.config.milestones_api_url, params)
|
|
|
|
if not isinstance(data, list):
|
|
raise GiteaError("Invalid response format: expected list of milestones")
|
|
|
|
return [self._parse_milestone(milestone_data) for milestone_data in data]
|
|
|
|
def create_milestone(self, milestone_data: MilestoneCreateData) -> Milestone:
|
|
"""Create a new milestone."""
|
|
payload = {
|
|
"title": milestone_data.title,
|
|
"description": milestone_data.description,
|
|
}
|
|
|
|
if milestone_data.due_on:
|
|
payload["due_on"] = milestone_data.due_on
|
|
|
|
data = self.http.post(self.config.milestones_api_url, payload)
|
|
return self._parse_milestone(data)
|
|
|
|
# Label operations
|
|
def list_labels(self) -> List[Label]:
|
|
"""List repository labels."""
|
|
data = self.http.get(self.config.labels_api_url)
|
|
|
|
if not isinstance(data, list):
|
|
raise GiteaError("Invalid response format: expected list of labels")
|
|
|
|
return [self._parse_label(label_data) for label_data in data]
|
|
|
|
def create_label(self, label_data: LabelCreateData) -> Label:
|
|
"""Create a new label."""
|
|
payload = {
|
|
"name": label_data.name,
|
|
"color": label_data.color,
|
|
"description": label_data.description,
|
|
}
|
|
|
|
data = self.http.post(self.config.labels_api_url, payload)
|
|
return self._parse_label(data)
|
|
|
|
# Parsing methods
|
|
def _parse_issue(self, data: Dict[str, Any]) -> Issue:
|
|
"""Parse issue data from API response."""
|
|
try:
|
|
# Parse labels
|
|
labels = []
|
|
if data.get('labels'):
|
|
labels = [self._parse_label(label_data) for label_data in data['labels']]
|
|
|
|
# Parse assignee
|
|
assignee = None
|
|
if data.get('assignee'):
|
|
assignee = self._parse_user(data['assignee'])
|
|
|
|
# Parse milestone
|
|
milestone = None
|
|
if data.get('milestone'):
|
|
milestone = self._parse_milestone(data['milestone'])
|
|
|
|
return Issue(
|
|
number=data['number'],
|
|
title=data['title'],
|
|
body=data.get('body', ''),
|
|
state=data['state'],
|
|
created_at=self._parse_datetime(data['created_at']),
|
|
updated_at=self._parse_datetime(data['updated_at']),
|
|
html_url=data['html_url'],
|
|
assignee=assignee,
|
|
labels=labels,
|
|
milestone=milestone
|
|
)
|
|
except (KeyError, ValueError) as e:
|
|
raise GiteaError(f"Failed to parse issue data: {e}")
|
|
|
|
def _parse_milestone(self, data: Dict[str, Any]) -> Milestone:
|
|
"""Parse milestone data from API response."""
|
|
return Milestone(
|
|
id=data['id'],
|
|
title=data['title'],
|
|
description=data.get('description', ''),
|
|
state=data['state'],
|
|
open_issues=data.get('open_issues', 0),
|
|
closed_issues=data.get('closed_issues', 0),
|
|
due_on=data.get('due_on'),
|
|
created_at=self._parse_datetime(data.get('created_at')) if data.get('created_at') else None,
|
|
updated_at=self._parse_datetime(data.get('updated_at')) if data.get('updated_at') else None
|
|
)
|
|
|
|
def _parse_label(self, data: Dict[str, Any]) -> Label:
|
|
"""Parse label data from API response."""
|
|
return Label(
|
|
id=data['id'],
|
|
name=data['name'],
|
|
color=data['color'],
|
|
description=data.get('description', '')
|
|
)
|
|
|
|
def _parse_user(self, data: Dict[str, Any]) -> User:
|
|
"""Parse user data from API response."""
|
|
return User(
|
|
id=data['id'],
|
|
login=data['login'],
|
|
full_name=data.get('full_name', ''),
|
|
email=data.get('email', ''),
|
|
avatar_url=data.get('avatar_url', '')
|
|
)
|
|
|
|
def _parse_datetime(self, date_str: str) -> datetime:
|
|
"""Parse datetime from API response."""
|
|
# Remove Z and microseconds for consistent parsing
|
|
date_str = date_str.replace('Z', '').split('.')[0]
|
|
return datetime.strptime(date_str, '%Y-%m-%dT%H:%M:%S') |