from typing import List, Optional from sqlalchemy.orm import Session from app.crud.base import CRUDBase from app.models.task import Task, TaskStatus from app.schemas.task import TaskCreate, TaskUpdate class CRUDTask(CRUDBase[Task, TaskCreate, TaskUpdate]): """CRUD operations for tasks.""" def get_multi_by_status( self, db: Session, *, status: TaskStatus, skip: int = 0, limit: int = 100 ) -> List[Task]: """Get multiple tasks by status. Args: db: Database session status: Task status skip: Number of records to skip limit: Maximum number of records to return Returns: List of tasks with specified status """ return ( db.query(self.model) .filter(self.model.status == status, not self.model.is_deleted) .offset(skip) .limit(limit) .all() ) def get_multi( self, db: Session, *, skip: int = 0, limit: int = 100 ) -> List[Task]: """Get multiple tasks, excluding deleted ones. Args: db: Database session skip: Number of records to skip limit: Maximum number of records to return Returns: List of non-deleted tasks """ return ( db.query(self.model) .filter(not self.model.is_deleted) .offset(skip) .limit(limit) .all() ) def get(self, db: Session, id: int) -> Optional[Task]: """Get a task by ID, excluding deleted ones. Args: db: Database session id: Task ID Returns: The task if found and not deleted, None otherwise """ return ( db.query(self.model) .filter(self.model.id == id, not self.model.is_deleted) .first() ) def remove(self, db: Session, *, id: int) -> Task: """Soft delete a task. Args: db: Database session id: Task ID Returns: The soft deleted task """ obj = db.query(self.model).get(id) if obj: obj.is_deleted = True db.add(obj) db.commit() db.refresh(obj) return obj task = CRUDTask(Task)