from sqlalchemy.orm import Session from app.dao.task_dao import TaskDAO from app.model.task_model import Task class TaskService: def __init__(self, db_session: Session): self.task_dao = TaskDAO(db_session) def get_all_tasks(self): return self.task_dao.get_all() def get_task_by_id(self, task_id: int): task = self.task_dao.get_by_id(task_id) if not task: raise ValueError(f"Task with id {task_id} not found.") return task def create_new_task(self, title: str, description: str = None): if not title: raise ValueError("Title cannot be empty.") new_task = Task(title=title, description=description) return self.task_dao.create(new_task) def update_existing_task(self, task_id: int, title: str = None, description: str = None, done: bool = None): task = self.get_task_by_id(task_id) # 复用 get_task_by_id 方法,包含了查找不到的逻辑 if title is not None: task.title = title if description is not None: task.description = description if done is not None: task.done = done return self.task_dao.update(task) def delete_task(self, task_id: int): task = self.get_task_by_id(task_id) self.task_dao.delete(task) return True