""" Gitea repository implementation with async HTTP client. Provides high-performance, reliable access to Gitea API with connection pooling, retry mechanisms, and proper error handling. """ import asyncio import json from infrastructure.logging import get_logger from typing import List, Optional, Dict, Any from datetime import datetime, timezone import aiohttp from domain.issues.models import Issue, Label, IssueState from domain.projects.models import Project, Milestone, ProjectState from infrastructure.repositories.interfaces import IssueRepository, ProjectRepository from infrastructure.connection_manager import ConnectionManager, retry_with_backoff, RetryConfig from infrastructure.exceptions import ( ErrorContext, OperationType, GiteaApiError, NetworkError, ResourceNotFoundError, ValidationError, ConcurrencyError ) logger = get_logger(__name__) class GiteaIssueRepository(IssueRepository): """ Gitea implementation of IssueRepository using async HTTP client. Provides efficient access to Gitea issues API with connection pooling, automatic retries, and proper error handling. """ def __init__(self, connection_manager: ConnectionManager, retry_config: Optional[RetryConfig] = None): self.connection_manager = connection_manager self.retry_config = retry_config or RetryConfig() self.repo_owner = None self.repo_name = None def set_repo_info(self, repo_owner: str, repo_name: str): """Set repository owner and name for API endpoints.""" self.repo_owner = repo_owner self.repo_name = repo_name @retry_with_backoff(RetryConfig()) async def get_issue(self, issue_number: int, context: Optional[ErrorContext] = None) -> Issue: """Retrieve an issue by its number from Gitea API.""" if context is None: context = ErrorContext( operation_id=f"get_issue_{issue_number}", operation_type=OperationType.READ, resource_type="Issue", resource_id=str(issue_number) ) try: session = await self.connection_manager.get_http_session() if not self.repo_owner or not self.repo_name: raise ValidationError("repo_info", None, "Repository owner and name must be set", context) endpoint = f"/api/v1/repos/{self.repo_owner}/{self.repo_name}/issues/{issue_number}" async with session.get(endpoint) as response: await self._handle_response_errors(response, context) data = await response.json() return self._map_api_issue_to_domain(data) except aiohttp.ClientError as e: logger.error(f"Network error getting issue {issue_number}: {e}") raise NetworkError(f"get issue {issue_number}", e, context) @retry_with_backoff(RetryConfig()) async def get_issues( self, project_id: Optional[str] = None, state: Optional[str] = None, labels: Optional[List[str]] = None, limit: int = 100, offset: int = 0, context: Optional[ErrorContext] = None ) -> List[Issue]: """Retrieve multiple issues with filtering and pagination.""" if context is None: context = ErrorContext( operation_id=f"get_issues_{project_id or 'all'}", operation_type=OperationType.READ, resource_type="Issue", metadata={ "project_id": project_id, "state": state, "labels": labels, "limit": limit, "offset": offset } ) try: session = await self.connection_manager.get_http_session() # Build query parameters params = { "limit": limit, "page": (offset // limit) + 1 # Gitea uses 1-based pagination } if state: params["state"] = state if labels: params["labels"] = ",".join(labels) if not self.repo_owner or not self.repo_name: raise ValidationError("repo_info", None, "Repository owner and name must be set", context) endpoint = f"/api/v1/repos/{self.repo_owner}/{self.repo_name}/issues" async with session.get(endpoint, params=params) as response: await self._handle_response_errors(response, context) data = await response.json() return [self._map_api_issue_to_domain(issue_data) for issue_data in data] except aiohttp.ClientError as e: logger.error(f"Network error getting issues: {e}") raise NetworkError("get issues", e, context) @retry_with_backoff(RetryConfig()) async def create_issue( self, title: str, body: str, labels: Optional[List[str]] = None, assignees: Optional[List[str]] = None, context: Optional[ErrorContext] = None ) -> Issue: """Create a new issue via Gitea API.""" if context is None: context = ErrorContext( operation_id=f"create_issue_{title[:50]}", operation_type=OperationType.WRITE, resource_type="Issue", request_data={ "title": title, "body": body, "labels": labels, "assignees": assignees } ) # Validate input if not title or not title.strip(): raise ValidationError("title", title, "Title cannot be empty", context) if len(title) > 255: raise ValidationError("title", title, "Title cannot exceed 255 characters", context) try: session = await self.connection_manager.get_http_session() # Prepare request payload payload = { "title": title.strip(), "body": body or "" } if labels: payload["labels"] = labels if assignees: payload["assignees"] = assignees if not self.repo_owner or not self.repo_name: raise ValidationError("repo_info", None, "Repository owner and name must be set", context) endpoint = f"/api/v1/repos/{self.repo_owner}/{self.repo_name}/issues" async with session.post(endpoint, json=payload) as response: await self._handle_response_errors(response, context) data = await response.json() created_issue = self._map_api_issue_to_domain(data) logger.info(f"Created issue #{created_issue.number}: {title}") return created_issue except aiohttp.ClientError as e: logger.error(f"Network error creating issue '{title}': {e}") raise NetworkError(f"create issue '{title}'", e, context) @retry_with_backoff(RetryConfig()) async def update_issue( self, issue_number: int, title: Optional[str] = None, body: Optional[str] = None, state: Optional[str] = None, labels: Optional[List[str]] = None, context: Optional[ErrorContext] = None ) -> Issue: """Update an existing issue via Gitea API.""" if context is None: context = ErrorContext( operation_id=f"update_issue_{issue_number}", operation_type=OperationType.UPDATE, resource_type="Issue", resource_id=str(issue_number), request_data={ "title": title, "body": body, "state": state, "labels": labels } ) # Validate input if title is not None: if not title.strip(): raise ValidationError("title", title, "Title cannot be empty", context) if len(title) > 255: raise ValidationError("title", title, "Title cannot exceed 255 characters", context) if state is not None and state not in ["open", "closed"]: raise ValidationError("state", state, "State must be 'open' or 'closed'", context) try: session = await self.connection_manager.get_http_session() # First, get current issue to check for concurrent modifications current_issue = await self.get_issue(issue_number, context) # Prepare update payload payload = {} if title is not None: payload["title"] = title.strip() if body is not None: payload["body"] = body if state is not None: payload["state"] = state if labels is not None: payload["labels"] = labels # Only update if there are changes if not payload: return current_issue if not self.repo_owner or not self.repo_name: raise ValidationError("repo_info", None, "Repository owner and name must be set", context) endpoint = f"/api/v1/repos/{self.repo_owner}/{self.repo_name}/issues/{issue_number}" async with session.patch(endpoint, json=payload) as response: # Handle potential concurrent modification if response.status == 409: raise ConcurrencyError("Issue", str(issue_number), context) await self._handle_response_errors(response, context) data = await response.json() updated_issue = self._map_api_issue_to_domain(data) logger.info(f"Updated issue #{issue_number}") return updated_issue except aiohttp.ClientError as e: logger.error(f"Network error updating issue {issue_number}: {e}") raise NetworkError(f"update issue {issue_number}", e, context) async def get_issue_project_info( self, issue_number: int, context: Optional[ErrorContext] = None ) -> Dict[str, Any]: """Get project-related information for an issue.""" if context is None: context = ErrorContext( operation_id=f"get_issue_project_info_{issue_number}", operation_type=OperationType.READ, resource_type="ProjectInfo", resource_id=str(issue_number) ) try: session = await self.connection_manager.get_http_session() # Get issue details first issue = await self.get_issue(issue_number, context) # Get repository information if not self.repo_owner or not self.repo_name: raise ValidationError("repo_info", None, "Repository owner and name must be set", context) repo_endpoint = f"/api/v1/repos/{self.repo_owner}/{self.repo_name}" async with session.get(repo_endpoint) as response: await self._handle_response_errors(response, context) repo_data = await response.json() # Get project boards if available project_info = { "repository": repo_data, "kanban_columns": ["Todo", "In Progress", "Review", "Done"], # Default columns "issue": { "number": issue.number, "title": issue.title, "state": issue.state.value, "labels": [label.name for label in issue.labels] } } # Try to get actual project boards try: projects_endpoint = f"/api/v1/repos/{self.repo_owner}/{self.repo_name}/projects" async with session.get(projects_endpoint) as projects_response: if projects_response.status == 200: projects_data = await projects_response.json() if projects_data: # Use first project's columns if available project_info["projects"] = projects_data except Exception: # Projects API might not be available, use defaults pass return project_info except aiohttp.ClientError as e: logger.error(f"Network error getting project info for issue {issue_number}: {e}") raise NetworkError(f"get project info for issue {issue_number}", e, context) def _map_api_issue_to_domain(self, api_data: Dict[str, Any]) -> Issue: """Map Gitea API issue data to domain Issue object.""" # Map labels labels = [] if "labels" in api_data: for label_data in api_data["labels"]: label = Label( name=label_data["name"], color=label_data.get("color", ""), description=label_data.get("description", "") ) labels.append(label) # Map state state_value = api_data.get("state", "open") issue_state = IssueState.OPEN if state_value == "open" else IssueState.CLOSED # Parse dates created_at = datetime.fromisoformat(api_data["created_at"].replace("Z", "+00:00")) updated_at = datetime.fromisoformat(api_data["updated_at"].replace("Z", "+00:00")) closed_at = None if api_data.get("closed_at"): closed_at = datetime.fromisoformat(api_data["closed_at"].replace("Z", "+00:00")) return Issue( number=api_data["number"], title=api_data["title"], state=issue_state, labels=labels, created_at=created_at, updated_at=updated_at, milestone=api_data.get("milestone", {}).get("title") if api_data.get("milestone") else None, assignee=api_data.get("assignees", [{}])[0].get("login") if api_data.get("assignees") else None, closed_at=closed_at ) async def _handle_response_errors(self, response: aiohttp.ClientResponse, context: ErrorContext): """Handle HTTP response errors and convert to appropriate exceptions.""" if response.status == 200 or response.status == 201: return response_text = await response.text() if response.status == 404: resource_id = context.resource_id or "unknown" raise ResourceNotFoundError(context.resource_type, resource_id, context) elif response.status == 401: raise GiteaApiError( response.status, "Authentication failed - check API token", str(response.url), context ) elif response.status == 403: raise GiteaApiError( response.status, "Access forbidden - check API permissions", str(response.url), context ) elif response.status == 409: # Conflict - usually concurrent modification raise ConcurrencyError(context.resource_type, context.resource_id or "unknown", context) elif response.status == 422: # Validation error try: error_data = await response.json() error_message = error_data.get("message", response_text) except: error_message = response_text raise ValidationError("request", None, error_message, context) elif response.status >= 500: raise GiteaApiError( response.status, f"Server error: {response_text}", str(response.url), context ) else: raise GiteaApiError( response.status, response_text, str(response.url), context ) class GiteaProjectRepository(ProjectRepository): """ Gitea implementation of ProjectRepository. Provides access to project and milestone information via Gitea API. """ def __init__(self, connection_manager: ConnectionManager, retry_config: Optional[RetryConfig] = None): self.connection_manager = connection_manager self.retry_config = retry_config or RetryConfig() @retry_with_backoff(RetryConfig()) async def get_project(self, project_id: str, context: Optional[ErrorContext] = None) -> Project: """Retrieve a project by its ID from Gitea API.""" if context is None: context = ErrorContext( operation_id=f"get_project_{project_id}", operation_type=OperationType.READ, resource_type="Project", resource_id=project_id ) try: session = await self.connection_manager.get_http_session() async with session.get(f"/api/v1/repos/projects/{project_id}") as response: await self._handle_response_errors(response, context) data = await response.json() return self._map_api_project_to_domain(data) except aiohttp.ClientError as e: logger.error(f"Network error getting project {project_id}: {e}") raise NetworkError(f"get project {project_id}", e, context) @retry_with_backoff(RetryConfig()) async def get_projects( self, organization: Optional[str] = None, limit: int = 100, offset: int = 0, context: Optional[ErrorContext] = None ) -> List[Project]: """Retrieve multiple projects with pagination.""" if context is None: context = ErrorContext( operation_id=f"get_projects_{organization or 'all'}", operation_type=OperationType.READ, resource_type="Project", metadata={ "organization": organization, "limit": limit, "offset": offset } ) try: session = await self.connection_manager.get_http_session() params = { "limit": limit, "page": (offset // limit) + 1 } endpoint = "/api/v1/repos/projects" if organization: endpoint = f"/api/v1/orgs/{organization}/projects" async with session.get(endpoint, params=params) as response: await self._handle_response_errors(response, context) data = await response.json() return [self._map_api_project_to_domain(project_data) for project_data in data] except aiohttp.ClientError as e: logger.error(f"Network error getting projects: {e}") raise NetworkError("get projects", e, context) @retry_with_backoff(RetryConfig()) async def get_milestones( self, project_id: str, state: Optional[str] = None, context: Optional[ErrorContext] = None ) -> List[Milestone]: """Retrieve milestones for a project.""" if context is None: context = ErrorContext( operation_id=f"get_milestones_{project_id}", operation_type=OperationType.READ, resource_type="Milestone", metadata={"project_id": project_id, "state": state} ) try: session = await self.connection_manager.get_http_session() params = {} if state: params["state"] = state # Note: This would need repo info from GiteaIssueRepository, but for now use general endpoint async with session.get("/api/v1/repos/milestones", params=params) as response: await self._handle_response_errors(response, context) data = await response.json() return [self._map_api_milestone_to_domain(milestone_data) for milestone_data in data] except aiohttp.ClientError as e: logger.error(f"Network error getting milestones for project {project_id}: {e}") raise NetworkError(f"get milestones for project {project_id}", e, context) @retry_with_backoff(RetryConfig()) async def create_milestone( self, project_id: str, title: str, description: str, due_date: Optional[str] = None, context: Optional[ErrorContext] = None ) -> Milestone: """Create a new milestone for a project.""" if context is None: context = ErrorContext( operation_id=f"create_milestone_{title[:50]}", operation_type=OperationType.WRITE, resource_type="Milestone", request_data={ "project_id": project_id, "title": title, "description": description, "due_date": due_date } ) # Validate input if not title or not title.strip(): raise ValidationError("title", title, "Milestone title cannot be empty", context) try: session = await self.connection_manager.get_http_session() payload = { "title": title.strip(), "description": description or "" } if due_date: payload["due_on"] = due_date # Note: This would need repo info from GiteaIssueRepository, but for now use general endpoint async with session.post("/api/v1/repos/milestones", json=payload) as response: await self._handle_response_errors(response, context) data = await response.json() created_milestone = self._map_api_milestone_to_domain(data) logger.info(f"Created milestone: {title}") return created_milestone except aiohttp.ClientError as e: logger.error(f"Network error creating milestone '{title}': {e}") raise NetworkError(f"create milestone '{title}'", e, context) def _map_api_project_to_domain(self, api_data: Dict[str, Any]) -> Project: """Map Gitea API project data to domain Project object.""" # For now, create a basic project since Gitea projects API might be limited created_at = datetime.fromisoformat(api_data.get("created_at", datetime.now(timezone.utc).isoformat()).replace("Z", "+00:00")) updated_at = datetime.fromisoformat(api_data.get("updated_at", datetime.now(timezone.utc).isoformat()).replace("Z", "+00:00")) return Project( id=str(api_data.get("id", 0)), name=api_data.get("title", api_data.get("name", "Unknown Project")), description=api_data.get("body", api_data.get("description", "")), state=ProjectState.ACTIVE, # Default to active milestones=[], # Will be populated separately created_at=created_at, updated_at=updated_at ) def _map_api_milestone_to_domain(self, api_data: Dict[str, Any]) -> Milestone: """Map Gitea API milestone data to domain Milestone object.""" created_at = datetime.fromisoformat(api_data["created_at"].replace("Z", "+00:00")) updated_at = datetime.fromisoformat(api_data["updated_at"].replace("Z", "+00:00")) due_date = None if api_data.get("due_on"): due_date = datetime.fromisoformat(api_data["due_on"].replace("Z", "+00:00")) return Milestone( id=api_data["id"], title=api_data["title"], description=api_data.get("description", ""), state=api_data.get("state", "open"), open_issues=api_data.get("open_issues", 0), closed_issues=api_data.get("closed_issues", 0), due_date=due_date, created_at=created_at, updated_at=updated_at ) async def _handle_response_errors(self, response: aiohttp.ClientResponse, context: ErrorContext): """Handle HTTP response errors and convert to appropriate exceptions.""" # Reuse the same error handling logic from GiteaIssueRepository if response.status == 200 or response.status == 201: return response_text = await response.text() if response.status == 404: resource_id = context.resource_id or "unknown" raise ResourceNotFoundError(context.resource_type, resource_id, context) elif response.status >= 400: raise GiteaApiError( response.status, response_text, str(response.url), context )