File size: 3,087 Bytes
1f2d50a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 |
"""Task service for KGraph-MCP API."""
import json
import logging
from pathlib import Path
from ..models.responses import TaskResponse
logger = logging.getLogger(__name__)
class TaskService:
"""Service for handling task-related operations."""
def __init__(self, tasks_file: str = "tasks.json"):
"""Initialize the task service."""
self.tasks_file = Path(tasks_file)
def _load_tasks(self) -> list[dict]:
"""Load tasks from JSON file."""
try:
if self.tasks_file.exists():
with open(self.tasks_file) as f:
return json.load(f)
return []
except Exception as e:
logger.error(f"Error loading tasks: {e}")
return []
def _save_tasks(self, tasks: list[dict]) -> bool:
"""Save tasks to JSON file."""
try:
with open(self.tasks_file, "w") as f:
json.dump(tasks, f, indent=2)
return True
except Exception as e:
logger.error(f"Error saving tasks: {e}")
return False
def get_all_tasks(self) -> list[TaskResponse]:
"""Get all tasks."""
tasks_data = self._load_tasks()
return [
TaskResponse(
id=task["id"],
title=task["title"],
description=task["description"],
status=task["status"],
dependencies=task.get("dependencies", []),
)
for task in tasks_data
]
def get_task_by_id(self, task_id: int) -> TaskResponse | None:
"""Get a task by ID."""
tasks_data = self._load_tasks()
for task in tasks_data:
if task["id"] == task_id:
return TaskResponse(
id=task["id"],
title=task["title"],
description=task["description"],
status=task["status"],
dependencies=task.get("dependencies", []),
)
return None
def create_task(
self, title: str, description: str, dependencies: list[int] = None
) -> TaskResponse:
"""Create a new task."""
if dependencies is None:
dependencies = []
tasks_data = self._load_tasks()
# Generate new ID
new_id = max([task["id"] for task in tasks_data], default=0) + 1
# Create new task
new_task = {
"id": new_id,
"title": title,
"description": description,
"status": "Todo",
"dependencies": dependencies,
"acceptance_criteria": [],
"created_at": "", # Will be set by taskmaster_mock.py
"updated_at": "",
"branch_name": None,
"pr_url": None,
}
tasks_data.append(new_task)
self._save_tasks(tasks_data)
return TaskResponse(
id=new_id,
title=title,
description=description,
status="Todo",
dependencies=dependencies,
)
|