File size: 9,437 Bytes
1941764 | 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 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 | """
Tasks API endpoints for CRUD operations on tasks.
This module provides:
- GET /api/tasks - List all user tasks
- POST /api/tasks - Create new task
- PUT /api/tasks/{id} - Update existing task
- DELETE /api/tasks/{id} - Delete task
All endpoints require JWT authentication and enforce user isolation.
"""
from fastapi import APIRouter, HTTPException, Depends, status
from sqlmodel import Session
from pydantic import BaseModel, Field
from typing import Optional, List
from ..models.task import Task
from ..services import tasks as task_service
from ..middleware.jwt_auth import get_current_user_id
from ..database import get_session
router = APIRouter()
# Request/Response Models
class CreateTaskRequest(BaseModel):
"""Request model for creating a task."""
title: str = Field(..., min_length=1, max_length=500, description="Task title")
description: Optional[str] = Field(None, description="Optional task description")
category: Optional[str] = Field(None, max_length=50, description="Task category/tag")
due_date: Optional[str] = Field(None, description="Due date in ISO format")
priority: Optional[str] = Field("medium", description="Task priority: low, medium, high")
is_recurring: Optional[bool] = Field(False, description="Whether task is recurring")
recurrence_type: Optional[str] = Field(None, description="Recurrence type: daily, weekly, monthly, yearly")
recurrence_interval: Optional[int] = Field(1, description="Recurrence interval (e.g., every 2 days)")
recurrence_end_date: Optional[str] = Field(None, description="Recurrence end date in ISO format")
class UpdateTaskRequest(BaseModel):
"""Request model for updating a task."""
title: Optional[str] = Field(None, min_length=1, max_length=500, description="Task title")
description: Optional[str] = Field(None, description="Task description")
completed: Optional[bool] = Field(None, description="Task completion status")
category: Optional[str] = Field(None, max_length=50, description="Task category/tag")
due_date: Optional[str] = Field(None, description="Due date in ISO format")
priority: Optional[str] = Field(None, description="Task priority: low, medium, high")
is_recurring: Optional[bool] = Field(None, description="Whether task is recurring")
recurrence_type: Optional[str] = Field(None, description="Recurrence type: daily, weekly, monthly, yearly")
recurrence_interval: Optional[int] = Field(None, description="Recurrence interval")
recurrence_end_date: Optional[str] = Field(None, description="Recurrence end date in ISO format")
class TaskResponse(BaseModel):
"""Task data response model."""
id: int
user_id: int
title: str
description: Optional[str]
completed: bool
category: Optional[str]
due_date: Optional[str]
priority: Optional[str]
is_recurring: bool
recurrence_type: Optional[str]
recurrence_interval: Optional[int]
recurrence_end_date: Optional[str]
parent_task_id: Optional[int]
created_at: str
updated_at: str
class TaskListResponse(BaseModel):
"""Response model for task list."""
tasks: List[TaskResponse]
@router.get("", response_model=TaskListResponse)
async def list_tasks(
user_id: int = Depends(get_current_user_id),
session: Session = Depends(get_session)
) -> TaskListResponse:
"""
Get all tasks for the authenticated user.
Args:
user_id: Current user ID from JWT token
session: Database session
Returns:
TaskListResponse with array of user's tasks
Raises:
HTTPException 401: If JWT token is invalid
"""
# Get user tasks
tasks = task_service.get_user_tasks(session, user_id)
# Convert to response format
task_responses = [
TaskResponse(
id=task.id,
user_id=task.user_id,
title=task.title,
description=task.description,
completed=task.completed,
category=task.category,
due_date=task.due_date.isoformat() if task.due_date else None,
priority=task.priority,
is_recurring=task.is_recurring,
recurrence_type=task.recurrence_type,
recurrence_interval=task.recurrence_interval,
recurrence_end_date=task.recurrence_end_date.isoformat() if task.recurrence_end_date else None,
parent_task_id=task.parent_task_id,
created_at=task.created_at.isoformat(),
updated_at=task.updated_at.isoformat()
)
for task in tasks
]
return TaskListResponse(tasks=task_responses)
@router.post("", response_model=TaskResponse, status_code=status.HTTP_201_CREATED)
async def create_task(
request: CreateTaskRequest,
user_id: int = Depends(get_current_user_id),
session: Session = Depends(get_session)
) -> TaskResponse:
"""
Create a new task for the authenticated user.
Args:
request: Task creation request with title and optional description
user_id: Current user ID from JWT token
session: Database session
Returns:
TaskResponse with created task data
Raises:
HTTPException 401: If JWT token is invalid
HTTPException 422: If validation fails
"""
# Create task
task = task_service.create_task(
session=session,
user_id=user_id,
title=request.title,
description=request.description,
category=request.category,
due_date=request.due_date,
priority=request.priority,
is_recurring=request.is_recurring or False,
recurrence_type=request.recurrence_type,
recurrence_interval=request.recurrence_interval or 1,
recurrence_end_date=request.recurrence_end_date
)
# Return response
return TaskResponse(
id=task.id,
user_id=task.user_id,
title=task.title,
description=task.description,
completed=task.completed,
category=task.category,
due_date=task.due_date.isoformat() if task.due_date else None,
priority=task.priority,
is_recurring=task.is_recurring,
recurrence_type=task.recurrence_type,
recurrence_interval=task.recurrence_interval,
recurrence_end_date=task.recurrence_end_date.isoformat() if task.recurrence_end_date else None,
parent_task_id=task.parent_task_id,
created_at=task.created_at.isoformat(),
updated_at=task.updated_at.isoformat()
)
@router.put("/{task_id}", response_model=TaskResponse)
async def update_task(
task_id: int,
request: UpdateTaskRequest,
user_id: int = Depends(get_current_user_id),
session: Session = Depends(get_session)
) -> TaskResponse:
"""
Update an existing task.
Args:
task_id: ID of the task to update
request: Task update request with optional fields
user_id: Current user ID from JWT token
session: Database session
Returns:
TaskResponse with updated task data
Raises:
HTTPException 401: If JWT token is invalid
HTTPException 404: If task not found or doesn't belong to user
"""
# Update task
task = task_service.update_task(
session=session,
task_id=task_id,
user_id=user_id,
title=request.title,
description=request.description,
completed=request.completed,
category=request.category,
due_date=request.due_date,
priority=request.priority,
is_recurring=request.is_recurring,
recurrence_type=request.recurrence_type,
recurrence_interval=request.recurrence_interval,
recurrence_end_date=request.recurrence_end_date
)
if not task:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Task not found"
)
# Return response
return TaskResponse(
id=task.id,
user_id=task.user_id,
title=task.title,
description=task.description,
completed=task.completed,
category=task.category,
due_date=task.due_date.isoformat() if task.due_date else None,
priority=task.priority,
is_recurring=task.is_recurring,
recurrence_type=task.recurrence_type,
recurrence_interval=task.recurrence_interval,
recurrence_end_date=task.recurrence_end_date.isoformat() if task.recurrence_end_date else None,
parent_task_id=task.parent_task_id,
created_at=task.created_at.isoformat(),
updated_at=task.updated_at.isoformat()
)
@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_task(
task_id: int,
user_id: int = Depends(get_current_user_id),
session: Session = Depends(get_session)
) -> None:
"""
Delete a task.
Args:
task_id: ID of the task to delete
user_id: Current user ID from JWT token
session: Database session
Returns:
None (204 No Content)
Raises:
HTTPException 401: If JWT token is invalid
HTTPException 404: If task not found or doesn't belong to user
"""
# Delete task
success = task_service.delete_task(
session=session,
task_id=task_id,
user_id=user_id
)
if not success:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Task not found"
)
|