Coverage for app/crud/tasks.py: 100.00%

1from uuid import UUID 

2 

3from sqlalchemy.orm import Session 

4 

5from app.models import Task, TaskStatus 

6from app.schemas import TaskCreate 

7 

8 

9def get_task(db: Session, task_id: UUID) -> Task | None: 

10 """Get task by ID""" 

11 return db.query(Task).filter(Task.taskId == task_id).first() 

12 

13 

14def get_tasks( 

15 db: Session, 

16 skip: int = 0, 

17 limit: int = 100, 

18 deployment_id: UUID | None = None, 

19 celery_task_id: str | None = None, 

20 status: TaskStatus | None = None 

21) -> list[Task]: 

22 """Get tasks with optional filters""" 

23 query = db.query(Task) 

24 if deployment_id: 

25 query = query.filter(Task.deploymentId == deployment_id) 

26 if celery_task_id: 

27 query = query.filter(Task.celeryTaskId == celery_task_id) 

28 if status: 

29 query = query.filter(Task.status == status) 

30 return query.offset(skip).limit(limit).all() 

31 

32 

33def create_task(db: Session, task: TaskCreate) -> Task: 

34 """Create a new task""" 

35 db_task = Task(**task) 

36 db.add(db_task) 

37 db.commit() 

38 db.refresh(db_task) 

39 return db_task 

40 

41 

42def update_task(db: Session, task_id: UUID, task_update) -> Task | None: 

43 """Update task information""" 

44 db_task = get_task(db, task_id) 

45 if not db_task: 

46 return None 

47 

48 # Handle both dict and Pydantic model 

49 if isinstance(task_update, dict): 

50 update_data = task_update 

51 else: 

52 update_data = task_update.model_dump(exclude_unset=True) 

53 

54 for field, value in update_data.items(): 

55 setattr(db_task, field, value) 

56 db.commit() 

57 db.refresh(db_task) 

58 return db_task 

59 

60 

61def delete_task(db: Session, task_id: UUID) -> bool: 

62 """Delete a task""" 

63 db_task = get_task(db, task_id) 

64 if not db_task: 

65 return False 

66 db.delete(db_task) 

67 db.commit() 

68 return True