Coverage for app / service.py: 97%

1from sqlalchemy.orm import Session 

2from app import models, schemas 

3from typing import Optional 

4 

5def get_task(db: Session, task_id: int): 

6 return db.query(models.Task).filter(models.Task.id == task_id).first() 

7 

8def get_tasks(db: Session, completed: Optional[bool] = None, priority: Optional[str] = None): 

9 q = db.query(models.Task) 

10 if completed is not None: 

11 q = q.filter(models.Task.completed == completed) 

12 if priority is not None: 

13 q = q.filter(models.Task.priority == priority) 

14 return q.all() 

15 

16def create_task(db: Session, task: schemas.TaskCreate): 

17 db_task = models.Task(title=task.title, description=task.description or "", priority=task.priority or "medium") 

18 db.add(db_task); db.commit(); db.refresh(db_task) 

19 return db_task 

20 

21def update_task(db: Session, task_id: int, task_data: schemas.TaskUpdate): 

22 db_task = get_task(db, task_id) 

23 if not db_task: return None 

24 for f, v in task_data.model_dump(exclude_unset=True).items(): 

25 setattr(db_task, f, v) 

26 db.commit(); db.refresh(db_task) 

27 return db_task 

28 

29def delete_task(db: Session, task_id: int): 

30 db_task = get_task(db, task_id) 

31 if not db_task: return False 

32 db.delete(db_task); db.commit() 

33 return True 

34 

35def get_task_stats(db: Session): 

36 total = db.query(models.Task).count() 

37 completed = db.query(models.Task).filter(models.Task.completed == True).count() 

38 return {"total": total, "completed": completed, "pending": total - completed} 

39 

40def mark_all_complete(db: Session): 

41 updated = db.query(models.Task).filter(models.Task.completed == False).update({"completed": True}) 

42 db.commit() 

43 return updated