""" Issue domain services. Contains business logic for issue-related operations. """ from typing import Dict, Any, List from datetime import datetime, timezone from .models import Issue, IssueState, LabelCategories from .exceptions import IssueValidationError class IssueStatusService: """Domain service for issue status-related business logic.""" def determine_kanban_column(self, issue: Issue, project_info: Dict[str, Any]) -> str: """Determine kanban column based on issue state and labels.""" # Pure business logic - no infrastructure dependencies label_categories = issue.categorize_labels() # Business rules for kanban column determination if issue.state == IssueState.CLOSED: return "Done" # Check for explicit status labels for state_label in label_categories.state_labels: if state_label == "status:in-progress": return "In Progress" elif state_label == "status:review": return "Review" elif state_label == "status:blocked": return "Blocked" elif state_label == "status:ready": return "Ready" # Default for open issues without explicit status return "Todo" def extract_priority_info(self, issue: Issue) -> Dict[str, Any]: """Extract priority information from issue labels.""" label_categories = issue.categorize_labels() priority_mapping = { "priority:low": "Low", "priority:medium": "Medium", "priority:high": "High", "priority:critical": "Critical" } for priority_label in label_categories.priority_labels: if priority_label in priority_mapping: return { "level": priority_mapping[priority_label], "label": priority_label } # Default priority return {"level": "Medium", "label": None} def extract_state_info(self, issue: Issue) -> Dict[str, Any]: """Extract state information from issue labels and state.""" label_categories = issue.categorize_labels() return { "state": issue.state.value, "state_labels": label_categories.state_labels, "is_closed": issue.state == IssueState.CLOSED, "closed_at": issue.closed_at.isoformat() if issue.closed_at else None } def calculate_issue_age_days(self, issue: Issue) -> int: """Calculate issue age in days.""" from datetime import datetime return (datetime.now(timezone.utc) - issue.created_at).days def is_stale_issue(self, issue: Issue, stale_threshold_days: int = 30) -> bool: """Determine if issue is considered stale based on business rules.""" if issue.state == IssueState.CLOSED: return False age_days = self.calculate_issue_age_days(issue) return age_days > stale_threshold_days class IssueValidationService: """Domain service for issue validation business rules.""" def validate_issue_creation(self, title: str, labels: List[str]) -> None: """Validate issue creation according to business rules.""" if not title or not title.strip(): raise IssueValidationError( "Issue title cannot be empty", field="title", value=title ) if len(title) > 255: raise IssueValidationError( "Issue title cannot exceed 255 characters", field="title", value=title ) # Business rule: Cannot have conflicting priority labels priority_labels = [label for label in labels if label.startswith("priority:")] if len(priority_labels) > 1: raise IssueValidationError( "Issue cannot have multiple priority labels", field="labels", value=priority_labels ) # Business rule: Cannot have conflicting state labels state_labels = [label for label in labels if label.startswith("status:")] if len(state_labels) > 1: raise IssueValidationError( "Issue cannot have multiple state labels", field="labels", value=state_labels ) def validate_title_update(self, new_title: str) -> None: """Validate issue title update.""" if not new_title or not new_title.strip(): raise IssueValidationError( "Issue title cannot be empty", field="title", value=new_title ) if len(new_title) > 255: raise IssueValidationError( "Issue title cannot exceed 255 characters", field="title", value=new_title ) def validate_label_addition(self, issue: Issue, new_label: str) -> None: """Validate adding a label to an issue.""" # Business rule: Cannot add duplicate labels if issue.has_label(new_label): raise IssueValidationError( f"Issue already has label '{new_label}'", field="labels", value=new_label ) # Business rule: Cannot add conflicting priority labels if new_label.startswith("priority:"): existing_priority_labels = [ label.name for label in issue.labels if label.is_priority_label() ] if existing_priority_labels: raise IssueValidationError( f"Issue already has priority label '{existing_priority_labels[0]}'. " f"Cannot add '{new_label}'", field="labels", value=new_label ) # Business rule: Cannot add conflicting state labels if new_label.startswith("status:"): existing_state_labels = [ label.name for label in issue.labels if label.is_state_label() ] if existing_state_labels: raise IssueValidationError( f"Issue already has state label '{existing_state_labels[0]}'. " f"Cannot add '{new_label}'", field="labels", value=new_label )