Spaces:
Sleeping
Sleeping
File size: 3,783 Bytes
e650b33 | 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 | from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List
from uuid import UUID
from datetime import datetime
from src.schemas.task import TaskResponse, TaskListResponse, TaskCreateRequest, TaskPatchRequest, TaskUpdateRequest
from src.models.task import Task
from src.models.user import User
from src.utils.deps import get_db, get_current_user
from src.utils.security import verify_token
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
router = APIRouter(tags=["Tasks"])
security = HTTPBearer()
@router.get("/{user_id}/tasks", response_model=TaskListResponse)
def get_user_tasks(
user_id: UUID,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
if user_id != current_user.id:
raise HTTPException(status_code=404, detail="Task not found")
tasks = (
db.query(Task)
.filter(Task.user_id == current_user.id)
.order_by(Task.created_at.desc())
.all()
)
return TaskListResponse(
tasks=tasks,
total=len(tasks),
)
@router.post("/{user_id}/tasks", response_model=TaskResponse, status_code=201)
def create_task(
user_id: UUID,
task_data: TaskCreateRequest,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
if user_id != current_user.id:
raise HTTPException(status_code=404, detail="User not found")
task = Task(
title=task_data.title,
description=task_data.description,
completed=task_data.completed,
user_id=current_user.id,
)
db.add(task)
db.commit()
db.refresh(task)
return task
@router.patch("/{user_id}/tasks/{task_id}", response_model=TaskResponse)
def update_task_partial(
user_id: UUID,
task_id: int,
task_update: TaskPatchRequest,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
if user_id != current_user.id:
raise HTTPException(status_code=404, detail="Task not found")
task = (
db.query(Task)
.filter(Task.id == task_id, Task.user_id == current_user.id)
.first()
)
if not task:
raise HTTPException(status_code=404, detail="Task not found")
if task_update.completed is not None:
task.completed = task_update.completed
db.commit()
db.refresh(task)
return task
@router.put("/{user_id}/tasks/{task_id}", response_model=TaskResponse)
def update_task_full(
user_id: UUID,
task_id: int,
task_update: TaskUpdateRequest,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
if user_id != current_user.id:
raise HTTPException(status_code=404, detail="Task not found")
task = (
db.query(Task)
.filter(Task.id == task_id, Task.user_id == current_user.id)
.first()
)
if not task:
raise HTTPException(status_code=404, detail="Task not found")
task.title = task_update.title
task.description = task_update.description
task.completed = task_update.completed
db.commit()
db.refresh(task)
return task
@router.delete("/{user_id}/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_task(
user_id: UUID,
task_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
if user_id != current_user.id:
raise HTTPException(status_code=404, detail="Task not found")
task = (
db.query(Task)
.filter(Task.id == task_id, Task.user_id == current_user.id)
.first()
)
if not task:
raise HTTPException(status_code=404, detail="Task not found")
db.delete(task)
db.commit()
|