programmingwithabdullah/todo-aipowered

1from typing import List2from sqlmodel import Session, select3from models.task import Task4from schemas.task_schemas import TaskCreate, TaskUpdate, TaskToggleComplete5from utils.errors import TaskNotFoundException, UnauthorizedTaskAccessException6 7 8class TaskService:9    @staticmethod10    def get_tasks_for_user(user_id: str, db: Session) -> List[Task]:11        statement = select(Task).where(Task.user_id == user_id)12        return list(db.exec(statement).all())13 14    @staticmethod15    def create_task(user_id: str, task: TaskCreate, db: Session) -> Task:16        db_task = Task(**task.model_dump(), user_id=user_id)17        db.add(db_task)18        db.commit()19        db.refresh(db_task)20        return db_task21 22    @staticmethod23    def get_task_by_id(user_id: str, task_id: int, db: Session) -> Task:24        task = db.get(Task, task_id)25        if not task:26            raise TaskNotFoundException(task_id)27 28        if task.user_id != user_id:29            raise UnauthorizedTaskAccessException()30 31        return task32 33    @staticmethod34    def get_tasks_by_title(user_id: str, title: str, db: Session) -> List[Task]:35        """36        Find tasks by title (case-insensitive partial match).37        Returns all matching tasks for disambiguation.38        """39        statement = select(Task).where(40            Task.user_id == user_id, Task.title.ilike(f"%{title}%")41        )42        return list(db.exec(statement).all())43 44    @staticmethod45    def update_task(46        user_id: str, task_id: int, task_update: TaskUpdate, db: Session47    ) -> Task:48        db_task = TaskService.get_task_by_id(user_id, task_id, db)49 50        # Update task fields51        update_data = task_update.model_dump(exclude_unset=True)52        for field, value in update_data.items():53            setattr(db_task, field, value)54 55        db.add(db_task)56        db.commit()57        db.refresh(db_task)58        return db_task59 60    @staticmethod61    def delete_task(user_id: str, task_id: int, db: Session) -> None:62        db_task = TaskService.get_task_by_id(user_id, task_id, db)63        db.delete(db_task)64        db.commit()65 66    @staticmethod67    def toggle_task_complete(68        user_id: str, task_id: int, task_toggle: TaskToggleComplete, db: Session69    ) -> Task:70        db_task = TaskService.get_task_by_id(user_id, task_id, db)71        db_task.completed = task_toggle.completed72        db.add(db_task)73        db.commit()74        db.refresh(db_task)75        return db_task76