refactor: Separate CLI presentation from core business logic
Complete architectural separation of concerns implementing clean layered design: • Services Layer: Pure business logic isolated from presentation - WorkspaceService: TDD workspace operations - IssueService: Issue management and creation - ProjectService: Project management and milestones - ExportService: Unix-friendly data export • CLI Layer: Clean presentation with command/presenter separation - Commands delegate to services for all business operations - Presenters handle formatted output and error messaging - Framework provides unified interface • Benefits: - Eliminates mixed concerns in 943-line CLI monolith - Enables easier testing and maintenance - Preserves all existing functionality and Unix pipeline compatibility - Provides foundation for future CLI development Resolves issue #20: CLI separation from core logic 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
28
services/__init__.py
Normal file
28
services/__init__.py
Normal file
@@ -0,0 +1,28 @@
|
||||
"""
|
||||
Business logic services layer.
|
||||
|
||||
This package contains pure business logic services that are independent of
|
||||
CLI presentation concerns. Services focus on:
|
||||
|
||||
- Core business operations
|
||||
- Data transformation
|
||||
- Validation and error handling
|
||||
- Integration with lower-level modules
|
||||
|
||||
Services should NOT:
|
||||
- Handle CLI arguments directly
|
||||
- Print output or format data for display
|
||||
- Call sys.exit() or handle CLI-specific errors
|
||||
"""
|
||||
|
||||
from .workspace_service import WorkspaceService
|
||||
from .issue_service import IssueService
|
||||
from .project_service import ProjectService
|
||||
from .export_service import ExportService
|
||||
|
||||
__all__ = [
|
||||
'WorkspaceService',
|
||||
'IssueService',
|
||||
'ProjectService',
|
||||
'ExportService'
|
||||
]
|
||||
150
services/export_service.py
Normal file
150
services/export_service.py
Normal file
@@ -0,0 +1,150 @@
|
||||
"""
|
||||
Export service - business logic for data export and formatting operations.
|
||||
"""
|
||||
|
||||
import json
|
||||
from typing import List, Dict, Any, Optional
|
||||
from datetime import datetime
|
||||
|
||||
from gitea.models import Issue
|
||||
from .issue_service import IssueService
|
||||
|
||||
|
||||
class ExportService:
|
||||
"""Service for export and data formatting operations."""
|
||||
|
||||
def __init__(self):
|
||||
self.issue_service = IssueService()
|
||||
|
||||
def get_issues_data(self, state: str = "all",
|
||||
sort_by: str = "number",
|
||||
filter_state: Optional[str] = None,
|
||||
filter_priority: Optional[str] = None,
|
||||
include_state: bool = False) -> List[Dict[str, Any]]:
|
||||
"""Get structured issue data for export.
|
||||
|
||||
Args:
|
||||
state: Issue state filter (all, open, closed)
|
||||
sort_by: Sort field (number, title, priority, state, created, updated)
|
||||
filter_state: Additional state filter (open, closed)
|
||||
filter_priority: Priority filter (low, medium, high, critical, none)
|
||||
include_state: Whether to include detailed state information
|
||||
|
||||
Returns:
|
||||
List of issue data dictionaries
|
||||
"""
|
||||
issues = self.issue_service.list_issues(state)
|
||||
|
||||
# Convert to structured data
|
||||
issue_data = []
|
||||
for issue in issues:
|
||||
# Get priority and state from labels
|
||||
priority = issue.priority or "none"
|
||||
status = issue.status or "none"
|
||||
|
||||
issue_info = {
|
||||
'number': issue.number,
|
||||
'title': issue.title.replace('\t', ' ').replace('\n', ' '), # Clean for TSV
|
||||
'priority': priority,
|
||||
'state': issue.state, # open/closed from basic data
|
||||
'status': status, # detailed status from labels
|
||||
'created': issue.created_at.strftime('%Y-%m-%d'),
|
||||
'updated': issue.updated_at.strftime('%Y-%m-%d')
|
||||
}
|
||||
issue_data.append(issue_info)
|
||||
|
||||
# Apply filters
|
||||
if filter_state:
|
||||
if filter_state == "open":
|
||||
issue_data = [i for i in issue_data if i['state'] == 'open']
|
||||
elif filter_state == "closed":
|
||||
issue_data = [i for i in issue_data if i['state'] == 'closed']
|
||||
|
||||
if filter_priority:
|
||||
issue_data = [i for i in issue_data if i['priority'] == filter_priority]
|
||||
|
||||
# Sort issues
|
||||
sort_key_map = {
|
||||
'number': lambda x: x['number'],
|
||||
'title': lambda x: x['title'].lower(),
|
||||
'priority': lambda x: {'critical': 4, 'high': 3, 'medium': 2, 'low': 1, 'none': 0}[x['priority']],
|
||||
'state': lambda x: x['state'],
|
||||
'created': lambda x: x['created'],
|
||||
'updated': lambda x: x['updated']
|
||||
}
|
||||
|
||||
if sort_by in sort_key_map:
|
||||
issue_data.sort(key=sort_key_map[sort_by], reverse=(sort_by in ['number', 'priority', 'created', 'updated']))
|
||||
|
||||
return issue_data
|
||||
|
||||
def format_issues_tsv(self, issue_data: List[Dict[str, Any]], include_state: bool = False) -> str:
|
||||
"""Format issues as TSV."""
|
||||
lines = []
|
||||
for issue in issue_data:
|
||||
if include_state:
|
||||
lines.append(f'{issue["number"]}\t{issue["title"]}\t{issue["priority"]}\t{issue["state"]}\t{issue["created"]}\t{issue["updated"]}')
|
||||
else:
|
||||
lines.append(f'{issue["number"]}\t{issue["title"]}\t{issue["priority"]}\t{issue["created"]}\t{issue["updated"]}')
|
||||
return '\n'.join(lines)
|
||||
|
||||
def format_issues_csv(self, issue_data: List[Dict[str, Any]], include_state: bool = False) -> str:
|
||||
"""Format issues as CSV."""
|
||||
lines = []
|
||||
|
||||
# Header
|
||||
if include_state:
|
||||
lines.append("number,title,priority,state,created,updated")
|
||||
for issue in issue_data:
|
||||
title = issue['title'].replace('"', '""') # Escape quotes
|
||||
lines.append(f'{issue["number"]},"{title}",{issue["priority"]},{issue["state"]},{issue["created"]},{issue["updated"]}')
|
||||
else:
|
||||
lines.append("number,title,priority,created,updated")
|
||||
for issue in issue_data:
|
||||
title = issue['title'].replace('"', '""')
|
||||
lines.append(f'{issue["number"]},"{title}",{issue["priority"]},{issue["created"]},{issue["updated"]}')
|
||||
|
||||
return '\n'.join(lines)
|
||||
|
||||
def format_issues_json(self, issue_data: List[Dict[str, Any]]) -> str:
|
||||
"""Format issues as JSON."""
|
||||
return json.dumps(issue_data, indent=2)
|
||||
|
||||
def format_issues_fields(self, issue_data: List[Dict[str, Any]], include_state: bool = False) -> str:
|
||||
"""Format issues as space-separated fields for awk processing."""
|
||||
lines = []
|
||||
|
||||
# Header
|
||||
if include_state:
|
||||
lines.append("NUMBER TITLE PRIORITY STATE CREATED UPDATED")
|
||||
for issue in issue_data:
|
||||
title = issue['title'].replace(' ', '_')
|
||||
lines.append(f'{issue["number"]} {title} {issue["priority"]} {issue["state"]} {issue["created"]} {issue["updated"]}')
|
||||
else:
|
||||
lines.append("NUMBER TITLE PRIORITY CREATED UPDATED")
|
||||
for issue in issue_data:
|
||||
title = issue['title'].replace(' ', '_')
|
||||
lines.append(f'{issue["number"]} {title} {issue["priority"]} {issue["created"]} {issue["updated"]}')
|
||||
|
||||
return '\n'.join(lines)
|
||||
|
||||
def export_issues(self, format_type: str = "tsv", **kwargs) -> str:
|
||||
"""Export issues in specified format.
|
||||
|
||||
Args:
|
||||
format_type: Output format (tsv, csv, json, fields)
|
||||
**kwargs: Export parameters (sort_by, filter_state, etc.)
|
||||
|
||||
Returns:
|
||||
Formatted string output
|
||||
"""
|
||||
issue_data = self.get_issues_data(**kwargs)
|
||||
|
||||
if format_type == "json":
|
||||
return self.format_issues_json(issue_data)
|
||||
elif format_type == "csv":
|
||||
return self.format_issues_csv(issue_data, kwargs.get('include_state', False))
|
||||
elif format_type == "fields":
|
||||
return self.format_issues_fields(issue_data, kwargs.get('include_state', False))
|
||||
else: # Default TSV
|
||||
return self.format_issues_tsv(issue_data, kwargs.get('include_state', False))
|
||||
69
services/issue_service.py
Normal file
69
services/issue_service.py
Normal file
@@ -0,0 +1,69 @@
|
||||
"""
|
||||
Issue service - business logic for issue operations.
|
||||
"""
|
||||
|
||||
from typing import List, Dict, Any, Optional
|
||||
from datetime import datetime
|
||||
|
||||
from tddai import IssueFetcher, TddaiError
|
||||
from tddai.issue_creator import IssueCreator
|
||||
from gitea.models import Issue, Priority
|
||||
|
||||
|
||||
class IssueService:
|
||||
"""Service for issue operations."""
|
||||
|
||||
def __init__(self):
|
||||
self.issue_fetcher = IssueFetcher()
|
||||
self.issue_creator = IssueCreator()
|
||||
|
||||
def get_issue(self, issue_number: int) -> Issue:
|
||||
"""Get a specific issue by number."""
|
||||
return self.issue_fetcher.fetch_issue(issue_number)
|
||||
|
||||
def list_issues(self, state: str = "all") -> List[Issue]:
|
||||
"""List issues with optional state filter."""
|
||||
return self.issue_fetcher.fetch_issues(state)
|
||||
|
||||
def list_open_issues(self) -> List[Issue]:
|
||||
"""List only open issues."""
|
||||
return self.issue_fetcher.fetch_open_issues()
|
||||
|
||||
def create_issue(self, title: str, body: str, **kwargs) -> Dict[str, Any]:
|
||||
"""Create a new issue."""
|
||||
return self.issue_creator.create_issue(title, body, **kwargs)
|
||||
|
||||
def create_enhancement_issue(self, title: str, use_case: str,
|
||||
technical_requirements: str = "",
|
||||
acceptance_criteria: List[str] = None,
|
||||
dependencies: List[str] = None,
|
||||
priority: str = "Medium") -> Dict[str, Any]:
|
||||
"""Create a structured enhancement issue."""
|
||||
return self.issue_creator.create_enhancement_issue(
|
||||
title, use_case, technical_requirements,
|
||||
acceptance_criteria, dependencies, priority
|
||||
)
|
||||
|
||||
def create_from_template(self, template_file: str, **kwargs) -> Dict[str, Any]:
|
||||
"""Create issue from template file."""
|
||||
return self.issue_creator.create_from_template(template_file, **kwargs)
|
||||
|
||||
def get_issue_summary(self, issue_number: int) -> Dict[str, Any]:
|
||||
"""Get issue summary for display purposes."""
|
||||
issue = self.get_issue(issue_number)
|
||||
|
||||
return {
|
||||
'number': issue.number,
|
||||
'title': issue.title,
|
||||
'body': issue.body,
|
||||
'state': issue.state,
|
||||
'priority': issue.priority,
|
||||
'status': issue.status,
|
||||
'created_at': issue.created_at,
|
||||
'updated_at': issue.updated_at,
|
||||
'html_url': issue.html_url,
|
||||
'assignee': issue.assignee.login if issue.assignee else None,
|
||||
'labels': [label.name for label in issue.labels],
|
||||
'has_milestone': issue.milestone is not None,
|
||||
'milestone_title': issue.milestone.title if issue.milestone else None
|
||||
}
|
||||
76
services/project_service.py
Normal file
76
services/project_service.py
Normal file
@@ -0,0 +1,76 @@
|
||||
"""
|
||||
Project service - business logic for project management operations.
|
||||
"""
|
||||
|
||||
from typing import List, Dict, Any
|
||||
|
||||
from tddai.project_manager import ProjectManager, ProjectState, Priority, Milestone, Label
|
||||
from tddai import TddaiError
|
||||
|
||||
|
||||
class ProjectService:
|
||||
"""Service for project management operations."""
|
||||
|
||||
def __init__(self):
|
||||
self.project_manager = ProjectManager()
|
||||
|
||||
def setup_project_management(self) -> None:
|
||||
"""Setup project management labels and structure."""
|
||||
self.project_manager.ensure_project_labels()
|
||||
|
||||
def create_milestone(self, title: str, description: str = "", due_date: str = None) -> Milestone:
|
||||
"""Create a new milestone (project)."""
|
||||
return self.project_manager.create_milestone(title, description, due_date)
|
||||
|
||||
def list_milestones(self, state: str = "open") -> List[Milestone]:
|
||||
"""List milestones."""
|
||||
return self.project_manager.list_milestones(state)
|
||||
|
||||
def list_labels(self) -> List[Label]:
|
||||
"""List repository labels."""
|
||||
return self.project_manager.list_labels()
|
||||
|
||||
def set_issue_state(self, issue_number: int, state_name: str) -> Dict[str, Any]:
|
||||
"""Set issue project state."""
|
||||
# Convert string to ProjectState enum
|
||||
state_map = {
|
||||
'todo': ProjectState.TODO,
|
||||
'active': ProjectState.ACTIVE,
|
||||
'review': ProjectState.REVIEW,
|
||||
'done': ProjectState.DONE,
|
||||
'blocked': ProjectState.BLOCKED
|
||||
}
|
||||
|
||||
if state_name not in state_map:
|
||||
raise TddaiError(f"Invalid state '{state_name}'. Valid states: {list(state_map.keys())}")
|
||||
|
||||
project_state = state_map[state_name]
|
||||
return self.project_manager.set_issue_state(issue_number, project_state)
|
||||
|
||||
def set_issue_priority(self, issue_number: int, priority_name: str) -> Dict[str, Any]:
|
||||
"""Set issue priority."""
|
||||
# Convert string to Priority enum
|
||||
priority_map = {
|
||||
'low': Priority.LOW,
|
||||
'medium': Priority.MEDIUM,
|
||||
'high': Priority.HIGH,
|
||||
'critical': Priority.CRITICAL
|
||||
}
|
||||
|
||||
if priority_name not in priority_map:
|
||||
raise TddaiError(f"Invalid priority '{priority_name}'. Valid priorities: {list(priority_map.keys())}")
|
||||
|
||||
priority_level = priority_map[priority_name]
|
||||
return self.project_manager.set_issue_priority(issue_number, priority_level)
|
||||
|
||||
def move_issue_to_done(self, issue_number: int) -> Dict[str, Any]:
|
||||
"""Move issue to done state and close it."""
|
||||
return self.project_manager.move_issue_to_done(issue_number)
|
||||
|
||||
def assign_issue_to_milestone(self, issue_number: int, milestone_id: int) -> Dict[str, Any]:
|
||||
"""Assign issue to a milestone."""
|
||||
return self.project_manager.assign_issue_to_milestone(issue_number, milestone_id)
|
||||
|
||||
def get_project_overview(self) -> Dict[str, Any]:
|
||||
"""Get project management overview."""
|
||||
return self.project_manager.get_project_overview()
|
||||
116
services/workspace_service.py
Normal file
116
services/workspace_service.py
Normal file
@@ -0,0 +1,116 @@
|
||||
"""
|
||||
Workspace service - business logic for TDD workspace operations.
|
||||
"""
|
||||
|
||||
from typing import Optional, Dict, Any
|
||||
from pathlib import Path
|
||||
|
||||
from tddai import WorkspaceManager, IssueFetcher, WorkspaceStatus, TddaiError
|
||||
|
||||
|
||||
class WorkspaceInfo:
|
||||
"""Value object for workspace information."""
|
||||
|
||||
def __init__(self, status: WorkspaceStatus, workspace=None):
|
||||
self.status = status
|
||||
self.workspace = workspace
|
||||
|
||||
@property
|
||||
def is_clean(self) -> bool:
|
||||
return self.status == WorkspaceStatus.CLEAN
|
||||
|
||||
@property
|
||||
def is_dirty(self) -> bool:
|
||||
return self.status == WorkspaceStatus.DIRTY
|
||||
|
||||
@property
|
||||
def is_active(self) -> bool:
|
||||
return self.status == WorkspaceStatus.ACTIVE
|
||||
|
||||
def get_test_files(self) -> list:
|
||||
"""Get list of test files in workspace."""
|
||||
if not self.workspace or not self.workspace.tests_dir.exists():
|
||||
return []
|
||||
return list(self.workspace.tests_dir.glob("*.py"))
|
||||
|
||||
|
||||
class WorkspaceService:
|
||||
"""Service for workspace operations."""
|
||||
|
||||
def __init__(self):
|
||||
self.workspace_manager = WorkspaceManager()
|
||||
self.issue_fetcher = IssueFetcher()
|
||||
|
||||
def get_workspace_info(self) -> WorkspaceInfo:
|
||||
"""Get current workspace information."""
|
||||
status = self.workspace_manager.get_status()
|
||||
workspace = None
|
||||
|
||||
if status == WorkspaceStatus.ACTIVE:
|
||||
workspace = self.workspace_manager.get_current_workspace()
|
||||
|
||||
return WorkspaceInfo(status, workspace)
|
||||
|
||||
def start_issue_workspace(self, issue_number: int) -> WorkspaceInfo:
|
||||
"""Start working on an issue.
|
||||
|
||||
Returns:
|
||||
WorkspaceInfo with the created workspace
|
||||
|
||||
Raises:
|
||||
TddaiError: If workspace already active or issue cannot be fetched
|
||||
"""
|
||||
# Check if workspace already active
|
||||
current_info = self.get_workspace_info()
|
||||
if current_info.is_active:
|
||||
raise TddaiError(f"Already working on issue #{current_info.workspace.issue_number}")
|
||||
|
||||
# Fetch issue data
|
||||
issue_data = self.issue_fetcher.get_issue_data_dict(issue_number)
|
||||
|
||||
# Create workspace
|
||||
workspace = self.workspace_manager.create_workspace(issue_data)
|
||||
return WorkspaceInfo(WorkspaceStatus.ACTIVE, workspace)
|
||||
|
||||
def finish_current_workspace(self) -> Optional[int]:
|
||||
"""Finish current workspace and return the issue number.
|
||||
|
||||
Returns:
|
||||
Issue number that was finished, or None if no active workspace
|
||||
|
||||
Raises:
|
||||
TddaiError: If workspace operations fail
|
||||
"""
|
||||
current_info = self.get_workspace_info()
|
||||
if not current_info.is_active:
|
||||
return None
|
||||
|
||||
issue_number = current_info.workspace.issue_number
|
||||
self.workspace_manager.finish_workspace()
|
||||
return issue_number
|
||||
|
||||
def get_workspace_summary(self) -> Dict[str, Any]:
|
||||
"""Get workspace summary for display purposes."""
|
||||
info = self.get_workspace_info()
|
||||
|
||||
summary = {
|
||||
'status': info.status,
|
||||
'active': info.is_active,
|
||||
'clean': info.is_clean,
|
||||
'dirty': info.is_dirty
|
||||
}
|
||||
|
||||
if info.workspace:
|
||||
test_files = info.get_test_files()
|
||||
summary.update({
|
||||
'issue_number': info.workspace.issue_number,
|
||||
'issue_title': info.workspace.issue_title,
|
||||
'issue_state': info.workspace.issue_state,
|
||||
'workspace_dir': str(info.workspace.workspace_dir),
|
||||
'test_count': len(test_files),
|
||||
'test_files': [f.name for f in test_files],
|
||||
'requirements_file': str(info.workspace.requirements_file),
|
||||
'test_plan_file': str(info.workspace.test_plan_file)
|
||||
})
|
||||
|
||||
return summary
|
||||
Reference in New Issue
Block a user