Spaces:
Sleeping
Sleeping
File size: 3,645 Bytes
f2b5c2a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 | """
Subtask service layer for business logic.
This module provides functions for:
- Creating subtasks
- Updating subtasks
- Deleting subtasks
- Getting subtasks for a task
"""
from sqlmodel import Session, select
from typing import List, Optional
from datetime import datetime
from ..models.subtask import Subtask
from ..models.task import Task
def get_task_subtasks(session: Session, task_id: int, user_id: int) -> List[Subtask]:
"""
Get all subtasks for a task (with user verification).
Args:
session: Database session
task_id: Task ID
user_id: User ID for verification
Returns:
List of subtasks ordered by order field
"""
# First verify the task belongs to the user
task = session.get(Task, task_id)
if not task or task.user_id != user_id:
return []
# Get subtasks
statement = select(Subtask).where(Subtask.task_id == task_id).order_by(Subtask.order)
subtasks = session.exec(statement).all()
return list(subtasks)
def create_subtask(
session: Session,
task_id: int,
user_id: int,
title: str,
order: int = 0
) -> Optional[Subtask]:
"""
Create a new subtask for a task.
Args:
session: Database session
task_id: Task ID
user_id: User ID for verification
title: Subtask title
order: Order position
Returns:
Created subtask or None if task doesn't belong to user
"""
# Verify task belongs to user
task = session.get(Task, task_id)
if not task or task.user_id != user_id:
return None
# Create subtask
subtask = Subtask(
task_id=task_id,
title=title,
order=order,
completed=False
)
session.add(subtask)
session.commit()
session.refresh(subtask)
return subtask
def update_subtask(
session: Session,
subtask_id: int,
user_id: int,
title: Optional[str] = None,
completed: Optional[bool] = None,
order: Optional[int] = None
) -> Optional[Subtask]:
"""
Update a subtask.
Args:
session: Database session
subtask_id: Subtask ID
user_id: User ID for verification
title: New title (optional)
completed: New completion status (optional)
order: New order (optional)
Returns:
Updated subtask or None if not found or doesn't belong to user
"""
# Get subtask and verify ownership through task
subtask = session.get(Subtask, subtask_id)
if not subtask:
return None
task = session.get(Task, subtask.task_id)
if not task or task.user_id != user_id:
return None
# Update fields
if title is not None:
subtask.title = title
if completed is not None:
subtask.completed = completed
if order is not None:
subtask.order = order
subtask.updated_at = datetime.utcnow()
session.add(subtask)
session.commit()
session.refresh(subtask)
return subtask
def delete_subtask(
session: Session,
subtask_id: int,
user_id: int
) -> bool:
"""
Delete a subtask.
Args:
session: Database session
subtask_id: Subtask ID
user_id: User ID for verification
Returns:
True if deleted, False if not found or doesn't belong to user
"""
# Get subtask and verify ownership through task
subtask = session.get(Subtask, subtask_id)
if not subtask:
return False
task = session.get(Task, subtask.task_id)
if not task or task.user_id != user_id:
return False
session.delete(subtask)
session.commit()
return True
|