Coverage for app/api/routes/tasks.py: 84%
168 statements
« prev ^ index » next coverage.py v7.6.1, created at 2026-03-24 11:40 +0000
1import uuid
2from datetime import datetime, timedelta
3from typing import Any
5from fastapi import APIRouter, Body, HTTPException
6from sqlmodel import col, func, select
8from app.api.deps import CurrentUser, SessionDep
9from app.models import Message, Task, TaskCreate, TaskPublic, TasksPublic, TaskUpdate
11router = APIRouter()
14@router.get("/", response_model=TasksPublic)
15def read_tasks(
16 session: SessionDep, current_user: CurrentUser, skip: int = 0, limit: int = 100
17) -> TasksPublic:
18 """
19 Retrieve tasks.
20 """
21 if current_user.is_superuser:
22 count_statement = select(func.count()).select_from(Task)
23 count = session.exec(count_statement).one()
24 statement = select(Task).order_by(col(Task.position)).offset(skip).limit(limit)
25 tasks = session.exec(statement).all()
26 else:
27 count_statement = (
28 select(func.count())
29 .select_from(Task)
30 .where(Task.owner_id == current_user.id)
31 )
32 count = session.exec(count_statement).one()
33 statement = (
34 select(Task)
35 .where(Task.owner_id == current_user.id)
36 .order_by(col(Task.position))
37 .offset(skip)
38 .limit(limit)
39 )
40 tasks = session.exec(statement).all()
42 return TasksPublic(data=tasks, count=count)
45@router.get("/{id}", response_model=TaskPublic)
46def read_task(session: SessionDep, current_user: CurrentUser, id: str) -> Task:
47 """
48 Get task by ID.
49 """
50 try:
51 task_id = uuid.UUID(id)
52 task = session.get(Task, task_id)
53 if not task:
54 raise HTTPException(status_code=404, detail="Task not found")
55 if not current_user.is_superuser and (task.owner_id != current_user.id):
56 raise HTTPException(status_code=400, detail="Not enough permissions")
57 return task
58 except ValueError:
59 raise HTTPException(status_code=422, detail="Invalid task ID format")
62@router.post("/", response_model=TaskPublic)
63def create_task(
64 *, session: SessionDep, current_user: CurrentUser, task_in: TaskCreate
65) -> Task:
66 """
67 Create new task.
68 """
69 # get the highest position value for the current user's tasks
70 max_position_query: Any = select(func.max(Task.position)).where(
71 Task.owner_id == current_user.id
72 )
73 max_position = session.exec(max_position_query).one() or 0
75 # set the new task's position to be one larger than the current max
76 task = Task.model_validate(
77 task_in, update={"owner_id": current_user.id, "position": max_position + 1}
78 )
79 session.add(task)
80 session.commit()
81 session.refresh(task)
82 return task
85@router.put("/{id}", response_model=TaskPublic)
86def update_task(
87 *, session: SessionDep, current_user: CurrentUser, id: str, task_in: TaskUpdate
88) -> Task:
89 """
90 Update a task.
91 """
92 try:
93 task_id = uuid.UUID(id)
94 task = session.get(Task, task_id)
95 if not task:
96 raise HTTPException(status_code=404, detail="Task not found")
97 if not current_user.is_superuser and (task.owner_id != current_user.id):
98 raise HTTPException(status_code=400, detail="Not enough permissions")
99 update_dict = task_in.model_dump(exclude_unset=True)
100 task.sqlmodel_update(update_dict)
101 session.add(task)
102 session.commit()
103 session.refresh(task)
104 return task
105 except ValueError:
106 raise HTTPException(status_code=422, detail="Invalid task ID format")
109@router.delete("/{id}")
110def delete_task(session: SessionDep, current_user: CurrentUser, id: str) -> Message:
111 """
112 Delete a task.
113 """
114 try:
115 task_id = uuid.UUID(id)
116 task = session.get(Task, task_id)
117 if not task:
118 raise HTTPException(status_code=404, detail="Task not found")
119 if not current_user.is_superuser and (task.owner_id != current_user.id):
120 raise HTTPException(status_code=400, detail="Not enough permissions")
121 session.delete(task)
122 session.commit()
123 return Message(message="Task deleted successfully")
124 except ValueError:
125 raise HTTPException(status_code=422, detail="Invalid task ID format")
128@router.post("/schedule", response_model=TasksPublic)
129def schedule_tasks(
130 session: SessionDep,
131 current_user: CurrentUser,
132 client_now: datetime | None = Body(
133 default=None, description="Client's current local time"
134 ),
135) -> TasksPublic:
136 """
137 Auto-schedule incomplete tasks with a due date into working-hour slots.
138 """
139 work_start = current_user.work_start
140 work_end = current_user.work_end
142 # Fetch incomplete tasks with a due date, ordered by priority ASC NULLS LAST, then due ASC
143 statement = (
144 select(Task)
145 .where(
146 Task.owner_id == current_user.id,
147 Task.completed == False, # noqa: E712
148 Task.due.is_not(None), # type: ignore[union-attr]
149 )
150 .order_by(
151 col(Task.priority_id).asc().nulls_last(),
152 col(Task.due).asc(),
153 )
154 )
155 tasks = list(session.exec(statement).all())
157 now = client_now if client_now is not None else datetime.utcnow()
158 # Round now up to the next 30-minute boundary
159 minutes = now.minute
160 remainder = minutes % 30
161 if remainder == 0 and now.second == 0 and now.microsecond == 0:
162 slot_start_base = now.replace(second=0, microsecond=0)
163 else:
164 slot_start_base = now.replace(second=0, microsecond=0) + timedelta(
165 minutes=(30 - remainder) % 30 if remainder != 0 else 0
166 )
167 # If remainder is 0 but there are seconds/microseconds, round up by 30
168 if remainder == 0:
169 slot_start_base = now.replace(second=0, microsecond=0) + timedelta(
170 minutes=30
171 )
173 # Track assigned slots as list of (start, end) tuples
174 assigned_slots: list[tuple[datetime, datetime]] = []
176 MAX_DAYS_AHEAD = 365
178 for task in tasks:
179 duration_minutes = task.duration if task.duration is not None else 30
180 duration = timedelta(minutes=duration_minutes)
181 due = task.due
183 candidate = slot_start_base
184 found_slot: datetime | None = None
186 # Iterate forward in 30-minute increments, up to MAX_DAYS_AHEAD days
187 limit = slot_start_base + timedelta(days=MAX_DAYS_AHEAD)
188 while candidate < limit:
189 candidate_end = candidate + duration
191 # Must end by the due date
192 if due is not None and candidate_end > due:
193 break
195 # Must be within working hours on this day
196 if candidate.hour < work_start or candidate.hour >= work_end:
197 # Jump to next working hour
198 if candidate.hour >= work_end:
199 # Move to next day at work_start
200 next_day = (candidate + timedelta(days=1)).replace(
201 hour=work_start, minute=0, second=0, microsecond=0
202 )
203 candidate = next_day
204 else:
205 # Before work_start today
206 today_start = candidate.replace(
207 hour=work_start, minute=0, second=0, microsecond=0
208 )
209 candidate = today_start
210 continue
212 # Also check that the slot end doesn't spill past work_end
213 work_end_dt = candidate.replace(
214 hour=work_end, minute=0, second=0, microsecond=0
215 )
216 if candidate_end > work_end_dt:
217 # Jump to next day at work_start
218 next_day = (candidate + timedelta(days=1)).replace(
219 hour=work_start, minute=0, second=0, microsecond=0
220 )
221 candidate = next_day
222 continue
224 # Check overlap with already-assigned slots
225 overlaps = False
226 for slot_s, slot_e in assigned_slots:
227 if candidate < slot_e and candidate_end > slot_s:
228 overlaps = True
229 # Jump to end of the overlapping slot
230 # Round up to next 30-min boundary after slot_e
231 slot_e_minutes = slot_e.minute % 30
232 if slot_e_minutes == 0 and slot_e.second == 0:
233 candidate = slot_e
234 else:
235 candidate = slot_e.replace(second=0, microsecond=0) + timedelta(
236 minutes=(30 - slot_e.minute % 30) % 30
237 )
238 if slot_e.minute % 30 == 0:
239 candidate = slot_e.replace(
240 second=0, microsecond=0
241 ) + timedelta(minutes=30)
242 break
243 if overlaps:
244 continue
246 # Valid slot found
247 found_slot = candidate
248 break
250 if found_slot is not None:
251 task.scheduled_start = found_slot
252 assigned_slots.append((found_slot, found_slot + duration))
253 session.add(task)
255 session.commit()
257 # Return full updated task list for this user
258 count_statement = (
259 select(func.count()).select_from(Task).where(Task.owner_id == current_user.id)
260 )
261 count = session.exec(count_statement).one()
262 all_tasks_statement = (
263 select(Task)
264 .where(Task.owner_id == current_user.id)
265 .order_by(col(Task.position))
266 )
267 all_tasks = session.exec(all_tasks_statement).all()
268 return TasksPublic(data=list(all_tasks), count=count)
271@router.post("/reorder", response_model=Message)
272def reorder_tasks(
273 *,
274 session: SessionDep,
275 current_user: CurrentUser,
276 task_ids: list[str] = Body(..., description="List of task IDs in the new order"),
277) -> Message:
278 """
279 Reorder tasks based on the provided list of task IDs.
280 """
281 # verify all tasks exist and belong to the current user
282 for position, task_id in enumerate(task_ids, start=1):
283 try:
284 task_uuid = uuid.UUID(task_id)
285 task = session.get(Task, task_uuid)
286 if not task:
287 raise HTTPException(
288 status_code=404, detail=f"Task with ID {task_id} not found"
289 )
290 if task.owner_id != current_user.id and not current_user.is_superuser:
291 raise HTTPException(status_code=400, detail="Not enough permissions")
293 # update the position
294 task.position = position
295 session.add(task)
296 except ValueError:
297 raise HTTPException(
298 status_code=422, detail=f"Invalid task ID format: {task_id}"
299 )
301 session.commit()
302 return Message(message="Tasks reordered successfully")
305@router.put("/{id}/toggle-completed", response_model=TaskPublic)
306def toggle_task_completed(
307 *, session: SessionDep, current_user: CurrentUser, id: str
308) -> Task:
309 """
310 Toggle the completed status of a task.
311 """
312 try:
313 task_id = uuid.UUID(id)
314 task = session.get(Task, task_id)
315 if not task:
316 raise HTTPException(status_code=404, detail="Task not found")
317 if not current_user.is_superuser and (task.owner_id != current_user.id):
318 raise HTTPException(status_code=400, detail="Not enough permissions")
320 task.completed = not task.completed
322 session.add(task)
323 session.commit()
324 session.refresh(task)
325 return task
326 except ValueError:
327 raise HTTPException(status_code=422, detail="Invalid task ID format")