contracts.py

• Pfad: /var/www/mcp-servers/mcp-tasks/domain/contracts.py
• Namespace: -
• Zeilen: 178 | Größe: 6,104 Bytes
• Geändert: 2025-12-28 12:55:51 | Gescannt: 2025-12-31 10:22:15

Code Hygiene Score: 89

• Dependencies: 70 (25%)
• LOC: 100 (20%)
• Methods: 100 (20%)
• Secrets: 100 (15%)
• Classes: 100 (10%)
• Magic Numbers: 60 (10%)

Issues 4

Dependencies 12

• extends str
• trait Enum
• trait Enum
• trait Enum
• use dataclasses.dataclass
• use dataclasses.field
• use datetime.datetime
• use enum.Enum
• use typing.Optional
• use typing.Dict
• use typing.Any
• use typing.List

Klassen 8

• TaskStatus class Zeile 8
• TaskType class Zeile 17
• ExecutorType class Zeile 24
• Task class Zeile 33
• TaskAssignment class Zeile 82
• TaskResult class Zeile 107
• TaskListResponse class Zeile 145
• QualityCheckResult class Zeile 162

Code

"""Domain Entities - Datenstrukturen für MCP-Tasks"""
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Optional, Dict, Any, List


class TaskStatus(str, Enum):
    """Task-Status-Werte"""
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"


class TaskType(str, Enum):
    """Task-Typen"""
    HUMAN_TASK = "human_task"
    AI_TASK = "ai_task"
    MIXED = "mixed"


class ExecutorType(str, Enum):
    """Executor-Typen für Zuweisungen und Ergebnisse"""
    HUMAN = "human"
    OLLAMA = "ollama"
    CLAUDE = "claude"
    ANTHROPIC_API = "anthropic_api"


@dataclass
class Task:
    """Task-Entity"""
    id: Optional[int] = None
    uuid: str = ""
    title: str = ""
    description: Optional[str] = None
    type: TaskType = TaskType.AI_TASK
    status: TaskStatus = TaskStatus.PENDING
    created_by: str = "mcp-tasks"
    created_by_type: str = "ai"
    parent_task_id: Optional[int] = None
    due_date: Optional[datetime] = None
    created_at: datetime = field(default_factory=datetime.now)
    updated_at: datetime = field(default_factory=datetime.now)
    completed_at: Optional[datetime] = None
    metadata: Dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> Dict[str, Any]:
        """Konvertiert zu Dictionary für JSON-Response (vollständig)"""
        return {
            "id": self.id,
            "uuid": self.uuid,
            "title": self.title,
            "description": self.description,
            "type": self.type.value if isinstance(self.type, TaskType) else self.type,
            "status": self.status.value if isinstance(self.status, TaskStatus) else self.status,
            "created_by": self.created_by,
            "created_by_type": self.created_by_type,
            "parent_task_id": self.parent_task_id,
            "due_date": self.due_date.isoformat() if self.due_date else None,
            "created_at": self.created_at.isoformat() if self.created_at else None,
            "updated_at": self.updated_at.isoformat() if self.updated_at else None,
            "completed_at": self.completed_at.isoformat() if self.completed_at else None,
            "metadata": self.metadata,
        }

    def to_dict_compact(self) -> Dict[str, Any]:
        """Kompakte Darstellung für Listen (Token-sparend)"""
        desc = self.description or ""
        return {
            "id": self.id,
            "title": self.title[:80] + "..." if len(self.title) > 80 else self.title,
            "description": desc[:100] + "..." if len(desc) > 100 else desc if desc else None,
            "type": self.type.value if isinstance(self.type, TaskType) else self.type,
            "status": self.status.value if isinstance(self.status, TaskStatus) else self.status,
        }


@dataclass
class TaskAssignment:
    """Task-Zuweisung"""
    id: Optional[int] = None
    task_id: int = 0
    assignee: str = ""
    assignee_type: ExecutorType = ExecutorType.HUMAN
    model_name: Optional[str] = None
    status: str = "pending"
    assigned_at: datetime = field(default_factory=datetime.now)
    notes: Optional[str] = None

    def to_dict(self) -> Dict[str, Any]:
        return {
            "id": self.id,
            "task_id": self.task_id,
            "assignee": self.assignee,
            "assignee_type": self.assignee_type.value if isinstance(self.assignee_type, ExecutorType) else self.assignee_type,
            "model_name": self.model_name,
            "status": self.status,
            "assigned_at": self.assigned_at.isoformat() if self.assigned_at else None,
            "notes": self.notes,
        }


@dataclass
class TaskResult:
    """Task-Ergebnis mit Token-Tracking"""
    id: Optional[int] = None
    task_id: int = 0
    executor: str = ""
    executor_type: ExecutorType = ExecutorType.OLLAMA
    model_name: Optional[str] = None
    request: Optional[str] = None
    response: Optional[str] = None
    status: str = "success"
    error_message: Optional[str] = None
    tokens_input: int = 0
    tokens_output: int = 0
    cost_usd: float = 0.0
    duration_ms: int = 0
    created_at: datetime = field(default_factory=datetime.now)

    def to_dict(self) -> Dict[str, Any]:
        return {
            "id": self.id,
            "task_id": self.task_id,
            "executor": self.executor,
            "executor_type": self.executor_type.value if isinstance(self.executor_type, ExecutorType) else self.executor_type,
            "model_name": self.model_name,
            "request": self.request[:500] + "..." if self.request and len(self.request) > 500 else self.request,
            "response": self.response[:1000] + "..." if self.response and len(self.response) > 1000 else self.response,
            "status": self.status,
            "error_message": self.error_message,
            "tokens_input": self.tokens_input,
            "tokens_output": self.tokens_output,
            "tokens_total": self.tokens_input + self.tokens_output,
            "cost_usd": self.cost_usd,
            "duration_ms": self.duration_ms,
            "created_at": self.created_at.isoformat() if self.created_at else None,
        }


@dataclass
class TaskListResponse:
    """Response für tasks_list"""
    tasks: List[Task] = field(default_factory=list)
    total: int = 0
    limit: int = 50
    offset: int = 0

    def to_dict(self) -> Dict[str, Any]:
        return {
            "tasks": [t.to_dict() for t in self.tasks],
            "total": self.total,
            "limit": self.limit,
            "offset": self.offset,
        }


@dataclass
class QualityCheckResult:
    """Ergebnis eines Quality-Checks"""
    check_name: str = ""
    passed: bool = True
    issues: int = 0
    fixed: int = 0
    details: Optional[str] = None

    def to_dict(self) -> Dict[str, Any]:
        return {
            "check_name": self.check_name,
            "passed": self.passed,
            "issues": self.issues,
            "fixed": self.fixed,
            "details": self.details,
        }