feat: Implement domain logic separation with clean architecture
- Created complete domain layer with pure business logic - Implemented Issue domain models with 48 passing tests - Implemented Project domain models with 31 passing tests - Added domain services for complex business operations - Established clean separation between domain, application, and infrastructure - All 250 tests passing with no breaking changes 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
162
domain/projects/models.py
Normal file
162
domain/projects/models.py
Normal file
@@ -0,0 +1,162 @@
|
||||
"""
|
||||
Project domain models.
|
||||
|
||||
Contains core business entities and value objects for project management.
|
||||
"""
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import List, Optional, Dict, Any
|
||||
from datetime import datetime
|
||||
from enum import Enum
|
||||
|
||||
from .exceptions import MilestoneError
|
||||
|
||||
|
||||
class ProjectState(Enum):
|
||||
"""Project state enumeration."""
|
||||
ACTIVE = "active"
|
||||
ARCHIVED = "archived"
|
||||
PLANNING = "planning"
|
||||
|
||||
|
||||
@dataclass
|
||||
class Milestone:
|
||||
"""Milestone entity."""
|
||||
id: int
|
||||
title: str
|
||||
description: Optional[str]
|
||||
due_date: Optional[datetime]
|
||||
state: str
|
||||
open_issues: int
|
||||
closed_issues: int
|
||||
|
||||
@property
|
||||
def completion_percentage(self) -> float:
|
||||
"""Calculate milestone completion percentage."""
|
||||
total_issues = self.open_issues + self.closed_issues
|
||||
if total_issues == 0:
|
||||
return 0.0
|
||||
return (self.closed_issues / total_issues) * 100
|
||||
|
||||
@property
|
||||
def total_issues(self) -> int:
|
||||
"""Get total number of issues in milestone."""
|
||||
return self.open_issues + self.closed_issues
|
||||
|
||||
def is_overdue(self) -> bool:
|
||||
"""Check if milestone is overdue."""
|
||||
if not self.due_date or self.state == "closed":
|
||||
return False
|
||||
return datetime.utcnow() > self.due_date
|
||||
|
||||
def is_completed(self) -> bool:
|
||||
"""Check if milestone is completed."""
|
||||
return self.state == "closed" or (self.total_issues > 0 and self.completion_percentage >= 100.0)
|
||||
|
||||
def add_issue(self) -> None:
|
||||
"""Add an open issue to the milestone."""
|
||||
self.open_issues += 1
|
||||
|
||||
def close_issue(self) -> None:
|
||||
"""Close an issue in the milestone."""
|
||||
if self.open_issues <= 0:
|
||||
raise MilestoneError(
|
||||
f"Cannot close issue in milestone '{self.title}': no open issues",
|
||||
milestone_id=self.id
|
||||
)
|
||||
self.open_issues -= 1
|
||||
self.closed_issues += 1
|
||||
|
||||
def reopen_issue(self) -> None:
|
||||
"""Reopen an issue in the milestone."""
|
||||
if self.closed_issues <= 0:
|
||||
raise MilestoneError(
|
||||
f"Cannot reopen issue in milestone '{self.title}': no closed issues",
|
||||
milestone_id=self.id
|
||||
)
|
||||
self.closed_issues -= 1
|
||||
self.open_issues += 1
|
||||
|
||||
|
||||
@dataclass
|
||||
class Project:
|
||||
"""Project aggregate root."""
|
||||
name: str
|
||||
description: str
|
||||
state: ProjectState
|
||||
milestones: List[Milestone]
|
||||
kanban_columns: List[str]
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
archived_at: Optional[datetime] = None
|
||||
|
||||
def get_active_milestones(self) -> List[Milestone]:
|
||||
"""Get milestones that are currently active."""
|
||||
return [milestone for milestone in self.milestones if milestone.state == "open"]
|
||||
|
||||
def get_completed_milestones(self) -> List[Milestone]:
|
||||
"""Get milestones that are completed."""
|
||||
return [milestone for milestone in self.milestones if milestone.is_completed()]
|
||||
|
||||
def get_overdue_milestones(self) -> List[Milestone]:
|
||||
"""Get milestones that are overdue."""
|
||||
return [milestone for milestone in self.milestones if milestone.is_overdue()]
|
||||
|
||||
def calculate_overall_progress(self) -> float:
|
||||
"""Calculate overall project progress based on milestones."""
|
||||
if not self.milestones:
|
||||
return 0.0
|
||||
|
||||
total_completion = sum(milestone.completion_percentage for milestone in self.milestones)
|
||||
return total_completion / len(self.milestones)
|
||||
|
||||
def get_total_issues(self) -> int:
|
||||
"""Get total number of issues across all milestones."""
|
||||
return sum(milestone.total_issues for milestone in self.milestones)
|
||||
|
||||
def get_total_open_issues(self) -> int:
|
||||
"""Get total number of open issues across all milestones."""
|
||||
return sum(milestone.open_issues for milestone in self.milestones)
|
||||
|
||||
def get_total_closed_issues(self) -> int:
|
||||
"""Get total number of closed issues across all milestones."""
|
||||
return sum(milestone.closed_issues for milestone in self.milestones)
|
||||
|
||||
def archive(self) -> None:
|
||||
"""Archive the project."""
|
||||
if self.state == ProjectState.ARCHIVED:
|
||||
return # Already archived
|
||||
|
||||
self.state = ProjectState.ARCHIVED
|
||||
self.archived_at = datetime.utcnow()
|
||||
|
||||
def activate(self) -> None:
|
||||
"""Activate the project."""
|
||||
if self.state == ProjectState.ACTIVE:
|
||||
return # Already active
|
||||
|
||||
self.state = ProjectState.ACTIVE
|
||||
self.archived_at = None
|
||||
|
||||
def add_milestone(self, milestone: Milestone) -> None:
|
||||
"""Add a milestone to the project."""
|
||||
# Check for duplicate milestone IDs
|
||||
if any(m.id == milestone.id for m in self.milestones):
|
||||
raise ValueError(f"Milestone with ID {milestone.id} already exists")
|
||||
|
||||
self.milestones.append(milestone)
|
||||
|
||||
def remove_milestone(self, milestone_id: int) -> None:
|
||||
"""Remove a milestone from the project."""
|
||||
original_count = len(self.milestones)
|
||||
self.milestones = [m for m in self.milestones if m.id != milestone_id]
|
||||
|
||||
if len(self.milestones) == original_count:
|
||||
raise ValueError(f"Milestone with ID {milestone_id} not found")
|
||||
|
||||
def get_milestone(self, milestone_id: int) -> Optional[Milestone]:
|
||||
"""Get a milestone by ID."""
|
||||
for milestone in self.milestones:
|
||||
if milestone.id == milestone_id:
|
||||
return milestone
|
||||
return None
|
||||
Reference in New Issue
Block a user