tegwick
1fa0f1e84a
Some checks failed
Test Suite / unit-tests (3.11) (push) Has been cancelled
Test Suite / unit-tests (3.12) (push) Has been cancelled
Test Suite / integration-tests (push) Has been cancelled
Test Suite / e2e-tests (push) Has been cancelled
Test Suite / performance-tests (push) Has been cancelled
Test Suite / code-quality (push) Has been cancelled
Test Suite / security-scan (push) Has been cancelled
Test Suite / test-summary (push) Has been cancelled
- Replace deprecated datetime.utcnow() with datetime.now(timezone.utc) across all domain models, services, infrastructure, and test files - Add missing timezone imports to all affected files - Fix pytest.ini configuration format from [tool:pytest] to [pytest] - Remove warning suppressions to expose actual issues - Ensure proper pytest marker registration for smoke tests Results: - 305 passed, 2 skipped, 0 warnings (down from 111 warnings) - All functionality preserved with modern datetime API usage - Improved code quality by addressing root causes vs suppression 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
189 lines
7.6 KiB
Python
189 lines
7.6 KiB
Python
"""
|
|
Project domain services.
|
|
|
|
Contains business logic for project-related operations.
|
|
"""
|
|
|
|
from typing import Dict, Any, List
|
|
from datetime import datetime, timedelta, timezone
|
|
from .models import Project, Milestone, ProjectState
|
|
from .exceptions import ProjectValidationError
|
|
|
|
|
|
class ProjectManagementService:
|
|
"""Domain service for project management business logic."""
|
|
|
|
def determine_project_health(self, project: Project) -> str:
|
|
"""Determine project health based on business rules."""
|
|
progress = project.calculate_overall_progress()
|
|
overdue_milestones = project.get_overdue_milestones()
|
|
active_milestones = project.get_active_milestones()
|
|
|
|
# Business rules for project health assessment
|
|
if project.state != ProjectState.ACTIVE:
|
|
return "Inactive"
|
|
|
|
if progress >= 95:
|
|
return "Excellent"
|
|
elif progress >= 80:
|
|
return "Good"
|
|
elif progress >= 60:
|
|
return "Fair"
|
|
elif len(overdue_milestones) > 0:
|
|
return "At Risk"
|
|
elif len(active_milestones) == 0:
|
|
return "Stalled"
|
|
else:
|
|
return "Needs Attention"
|
|
|
|
def calculate_project_velocity(self, project: Project, days_back: int = 30) -> float:
|
|
"""Calculate project velocity based on recent milestone completions."""
|
|
completed_milestones = project.get_completed_milestones()
|
|
cutoff_date = datetime.now(timezone.utc) - timedelta(days=days_back)
|
|
|
|
# Count milestones completed in the specified period
|
|
# Note: This would need milestone completion dates in a real implementation
|
|
recent_completions = len(completed_milestones) # Simplified for now
|
|
|
|
return recent_completions / (days_back / 7) # Issues per week
|
|
|
|
def identify_bottlenecks(self, project: Project) -> List[str]:
|
|
"""Identify potential bottlenecks in the project."""
|
|
bottlenecks = []
|
|
|
|
# Check for overdue milestones
|
|
overdue_milestones = project.get_overdue_milestones()
|
|
if overdue_milestones:
|
|
bottlenecks.append(f"Overdue milestones: {len(overdue_milestones)}")
|
|
|
|
# Check for milestones with too many open issues
|
|
for milestone in project.get_active_milestones():
|
|
if milestone.open_issues > 20: # Business rule: threshold for too many issues
|
|
bottlenecks.append(f"Milestone '{milestone.title}' has {milestone.open_issues} open issues")
|
|
|
|
# Check for stalled milestones (no progress)
|
|
for milestone in project.get_active_milestones():
|
|
if milestone.total_issues > 0 and milestone.completion_percentage == 0:
|
|
bottlenecks.append(f"Milestone '{milestone.title}' shows no progress")
|
|
|
|
return bottlenecks
|
|
|
|
def recommend_next_actions(self, project: Project) -> List[str]:
|
|
"""Recommend next actions based on project state."""
|
|
recommendations = []
|
|
health = self.determine_project_health(project)
|
|
|
|
if health == "At Risk":
|
|
overdue_milestones = project.get_overdue_milestones()
|
|
recommendations.append(f"Address {len(overdue_milestones)} overdue milestone(s)")
|
|
|
|
if health == "Stalled":
|
|
recommendations.append("Create new milestones or reactivate existing ones")
|
|
|
|
# Check for milestones nearing completion
|
|
for milestone in project.get_active_milestones():
|
|
if milestone.completion_percentage >= 80:
|
|
recommendations.append(f"Focus on completing milestone '{milestone.title}' ({milestone.completion_percentage:.0f}% done)")
|
|
|
|
# Check for unbalanced workload
|
|
total_open = project.get_total_open_issues()
|
|
if total_open > 50: # Business rule: threshold for too many open issues
|
|
recommendations.append(f"Consider breaking down work - {total_open} total open issues")
|
|
|
|
return recommendations
|
|
|
|
def validate_project_creation(self, name: str, description: str) -> None:
|
|
"""Validate project creation according to business rules."""
|
|
if not name or not name.strip():
|
|
raise ProjectValidationError(
|
|
"Project name cannot be empty",
|
|
field="name",
|
|
value=name
|
|
)
|
|
|
|
if len(name) > 100:
|
|
raise ProjectValidationError(
|
|
"Project name cannot exceed 100 characters",
|
|
field="name",
|
|
value=name
|
|
)
|
|
|
|
if description and len(description) > 1000:
|
|
raise ProjectValidationError(
|
|
"Project description cannot exceed 1000 characters",
|
|
field="description",
|
|
value=description
|
|
)
|
|
|
|
def validate_milestone_creation(self, title: str, due_date: datetime = None) -> None:
|
|
"""Validate milestone creation according to business rules."""
|
|
if not title or not title.strip():
|
|
raise ProjectValidationError(
|
|
"Milestone title cannot be empty",
|
|
field="title",
|
|
value=title
|
|
)
|
|
|
|
if len(title) > 100:
|
|
raise ProjectValidationError(
|
|
"Milestone title cannot exceed 100 characters",
|
|
field="title",
|
|
value=title
|
|
)
|
|
|
|
# Business rule: Due date cannot be in the past
|
|
if due_date and due_date < datetime.now(timezone.utc):
|
|
raise ProjectValidationError(
|
|
"Milestone due date cannot be in the past",
|
|
field="due_date",
|
|
value=due_date
|
|
)
|
|
|
|
def calculate_milestone_priority(self, milestone: Milestone) -> int:
|
|
"""Calculate milestone priority based on business rules."""
|
|
priority_score = 0
|
|
|
|
# Higher priority for milestones with more issues
|
|
priority_score += milestone.total_issues * 2
|
|
|
|
# Higher priority for milestones with due dates
|
|
if milestone.due_date:
|
|
days_until_due = (milestone.due_date - datetime.now(timezone.utc)).days
|
|
if days_until_due <= 7:
|
|
priority_score += 50 # Very urgent
|
|
elif days_until_due <= 30:
|
|
priority_score += 25 # Urgent
|
|
else:
|
|
priority_score += 10 # Normal
|
|
|
|
# Higher priority for milestones closer to completion
|
|
if milestone.completion_percentage >= 75:
|
|
priority_score += 30 # Push to completion
|
|
|
|
# Lower priority for stalled milestones
|
|
if milestone.total_issues > 0 and milestone.completion_percentage == 0:
|
|
priority_score -= 20
|
|
|
|
return max(0, priority_score) # Ensure non-negative
|
|
|
|
def generate_project_summary(self, project: Project) -> Dict[str, Any]:
|
|
"""Generate a comprehensive project summary."""
|
|
health = self.determine_project_health(project)
|
|
bottlenecks = self.identify_bottlenecks(project)
|
|
recommendations = self.recommend_next_actions(project)
|
|
|
|
return {
|
|
"name": project.name,
|
|
"state": project.state.value,
|
|
"health": health,
|
|
"overall_progress": project.calculate_overall_progress(),
|
|
"total_milestones": len(project.milestones),
|
|
"active_milestones": len(project.get_active_milestones()),
|
|
"completed_milestones": len(project.get_completed_milestones()),
|
|
"overdue_milestones": len(project.get_overdue_milestones()),
|
|
"total_issues": project.get_total_issues(),
|
|
"open_issues": project.get_total_open_issues(),
|
|
"closed_issues": project.get_total_closed_issues(),
|
|
"bottlenecks": bottlenecks,
|
|
"recommendations": recommendations
|
|
} |