tegwick
fb968dff34
- Fix issue manager to properly read API token and repo info from main MarkiTect config - Update Gitea plugin to use correct repository-specific API endpoints - Correct domain model mapping to only include valid Issue model fields - Fix presentation layer to safely access optional body attribute - Enable full functionality for 'markitect issues show' and 'markitect issues list' commands 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
102 lines
4.2 KiB
Python
102 lines
4.2 KiB
Python
"""
|
|
Gitea backend plugin for issue management.
|
|
|
|
This plugin integrates with existing GiteaIssueRepository infrastructure.
|
|
"""
|
|
|
|
import asyncio
|
|
from typing import List, Optional, Dict, Any
|
|
|
|
from ..base import IssueBackend
|
|
from domain.issues.models import Issue
|
|
from infrastructure.repositories.gitea_repository import GiteaIssueRepository
|
|
from infrastructure.connection_manager import ConnectionManager, DataSourceConfig
|
|
|
|
|
|
class GiteaPlugin(IssueBackend):
|
|
"""Gitea backend plugin using existing repository infrastructure."""
|
|
|
|
def __init__(self, config: Dict[str, Any]):
|
|
"""Initialize Gitea plugin with configuration."""
|
|
super().__init__(config)
|
|
|
|
# Store repo info for API endpoints
|
|
self.repo_owner = config.get('repo_owner', 'coulomb')
|
|
self.repo_name = config.get('repo_name', 'markitect_project')
|
|
self.repo_full_name = f"{self.repo_owner}/{self.repo_name}"
|
|
|
|
# Create connection manager with configuration
|
|
datasource_config = DataSourceConfig(
|
|
gitea_base_url=config.get('url', 'http://92.205.130.254:32166'),
|
|
gitea_token=config.get('token', ''),
|
|
database_path=config.get('database_path', 'markitect.db')
|
|
)
|
|
connection_manager = ConnectionManager(datasource_config)
|
|
|
|
# Create repository with repo info
|
|
self.repository = GiteaIssueRepository(connection_manager)
|
|
self.repository.set_repo_info(self.repo_owner, self.repo_name)
|
|
|
|
def list_issues(self, state: Optional[str] = None) -> List[Issue]:
|
|
"""List issues from Gitea."""
|
|
return asyncio.run(self._list_issues_async(state))
|
|
|
|
async def _list_issues_async(self, state: Optional[str] = None) -> List[Issue]:
|
|
"""Async implementation of list_issues."""
|
|
if state == 'all' or state is None:
|
|
state = None # Repository expects None for all issues
|
|
return await self.repository.get_issues(state=state)
|
|
|
|
def get_issue(self, issue_id: str) -> Issue:
|
|
"""Get specific issue from Gitea."""
|
|
return asyncio.run(self._get_issue_async(issue_id))
|
|
|
|
async def _get_issue_async(self, issue_id: str) -> Issue:
|
|
"""Async implementation of get_issue."""
|
|
issue_number = int(issue_id)
|
|
return await self.repository.get_issue(issue_number)
|
|
|
|
def create_issue(self, title: str, body: str, **kwargs) -> Issue:
|
|
"""Create new issue in Gitea."""
|
|
return asyncio.run(self._create_issue_async(title, body, **kwargs))
|
|
|
|
async def _create_issue_async(self, title: str, body: str, **kwargs) -> Issue:
|
|
"""Async implementation of create_issue."""
|
|
return await self.repository.create_issue(title=title, body=body, **kwargs)
|
|
|
|
def add_comment(self, issue_id: str, comment: str) -> Dict[str, Any]:
|
|
"""Add comment to Gitea issue."""
|
|
return asyncio.run(self._add_comment_async(issue_id, comment))
|
|
|
|
async def _add_comment_async(self, issue_id: str, comment: str) -> Dict[str, Any]:
|
|
"""Async implementation of add_comment."""
|
|
if not comment.strip():
|
|
raise ValueError("Comment cannot be empty")
|
|
if not issue_id.strip():
|
|
raise ValueError("Issue ID cannot be empty")
|
|
|
|
# For now, return mock comment data
|
|
# This will be implemented when comment support is added to repository
|
|
return {
|
|
'id': 'comment_123',
|
|
'body': comment,
|
|
'issue_id': issue_id
|
|
}
|
|
|
|
def close_issue(self, issue_id: str) -> Issue:
|
|
"""Close issue in Gitea."""
|
|
return asyncio.run(self._close_issue_async(issue_id))
|
|
|
|
async def _close_issue_async(self, issue_id: str) -> Issue:
|
|
"""Async implementation of close_issue."""
|
|
issue_number = int(issue_id)
|
|
return await self.repository.update_issue(issue_number, state='closed')
|
|
|
|
def update_issue(self, issue_id: str, **kwargs) -> Issue:
|
|
"""Update issue in Gitea."""
|
|
return asyncio.run(self._update_issue_async(issue_id, **kwargs))
|
|
|
|
async def _update_issue_async(self, issue_id: str, **kwargs) -> Issue:
|
|
"""Async implementation of update_issue."""
|
|
issue_number = int(issue_id)
|
|
return await self.repository.update_issue(issue_number, **kwargs) |