File size: 4,590 Bytes
67f8819 9e95910 67f8819 | 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 | """
Task CRUD service with user isolation.
Per @specs/001-auth-api-bridge/data-model.md and @specs/001-auth-api-bridge/api/rest-endpoints.md
"""
from sqlmodel import Session, select
from typing import List, Optional
from datetime import datetime
from uuid import UUID
from models.task import TaskTable
from config import engine
class TaskService:
"""Service for task CRUD operations with enforced user isolation."""
@staticmethod
def get_user_tasks(session: Session, user_id: UUID) -> List[TaskTable]:
"""
Retrieve all tasks for a specific user.
Per @specs/001-auth-api-bridge/data-model.md query patterns
"""
statement = select(TaskTable).where(TaskTable.user_id == user_id)
return session.exec(statement).all()
@staticmethod
def get_task_by_id(session: Session, task_id: UUID, user_id: UUID) -> Optional[TaskTable]:
"""
Retrieve a task if it belongs to the user.
Critical for security - enforces user ownership check.
Per @specs/001-auth-api-bridge/data-model.md
"""
return session.query(TaskTable).filter(
TaskTable.id == task_id,
TaskTable.user_id == user_id # Critical for security
).first()
@staticmethod
def create_task(session: Session, user_id: UUID, title: str, description: Optional[str], priority: str = "medium") -> TaskTable:
"""
Create a new task for the user.
Args:
session: Database session
user_id: Owner's user ID (from JWT)
title: Task title
description: Optional task description
priority: Task priority level (low, medium, high) - defaults to medium
Returns:
Created task with user_id set
"""
task = TaskTable(
user_id=user_id, # From verified JWT
title=title,
description=description,
priority=priority,
completed=False,
created_at=datetime.utcnow()
)
session.add(task)
session.commit()
session.refresh(task)
return task
@staticmethod
def complete_task(session: Session, task_id: UUID, user_id: UUID) -> Optional[TaskTable]:
"""
Mark a task as completed.
Args:
session: Database session
task_id: Task to complete
user_id: User requesting completion (for ownership verification)
Returns:
Updated task if found and owned by user, None otherwise
"""
task = TaskService.get_task_by_id(session, task_id, user_id)
if task:
task.completed = True
task.completed_at = datetime.utcnow()
session.add(task)
session.commit()
session.refresh(task)
return task
@staticmethod
def delete_task(session: Session, task_id: UUID, user_id: UUID) -> bool:
"""
Delete a task if it belongs to the user.
Args:
session: Database session
task_id: Task to delete
user_id: User requesting deletion (for ownership verification)
Returns:
True if deleted, False if not found
"""
task = TaskService.get_task_by_id(session, task_id, user_id)
if task:
session.delete(task)
session.commit()
return True
return False
@staticmethod
def update_task(session: Session, task_id: UUID, user_id: UUID, title: Optional[str] = None, description: Optional[str] = None, priority: Optional[str] = None) -> Optional[TaskTable]:
"""
Update a task's title, description, and/or priority if it belongs to the user.
Args:
session: Database session
task_id: Task to update
user_id: User requesting update (for ownership verification)
title: New title (optional)
description: New description (optional)
priority: New priority level (optional)
Returns:
Updated task if found and owned by user, None otherwise
"""
task = TaskService.get_task_by_id(session, task_id, user_id)
if task:
if title is not None:
task.title = title
if description is not None:
task.description = description
if priority is not None:
task.priority = priority
session.add(task)
session.commit()
session.refresh(task)
return task
|