Spaces:
Sleeping
Sleeping
File size: 3,321 Bytes
a291087 | 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 | from fastapi import APIRouter, Depends, HTTPException, status
from sqlmodel import Session, select
from typing import List
from datetime import datetime
from ..api.dependencies import get_session, CurrentUserDep
from ..models.task import Task, TaskCreate, TaskUpdate, TaskRead
router = APIRouter(prefix="/api/tasks", tags=["tasks"])
@router.get("", response_model=List[TaskRead])
def list_tasks(
current_user: CurrentUserDep,
session: Session = Depends(get_session)
):
statement = select(Task).where(Task.user_id == current_user.id).order_by(Task.completed.asc(), Task.created_at.desc())
tasks = session.execute(statement).scalars().all()
return tasks
@router.post("", response_model=TaskRead, status_code=status.HTTP_201_CREATED)
def create_task(
current_user: CurrentUserDep,
task_in: TaskCreate,
session: Session = Depends(get_session)
):
db_task = Task.model_validate(task_in, update={"user_id": current_user.id})
session.add(db_task)
session.commit()
session.refresh(db_task)
return db_task
@router.get("/{task_id}", response_model=TaskRead)
def read_task(
current_user: CurrentUserDep,
task_id: int,
session: Session = Depends(get_session)
):
task = session.get(Task, task_id)
if not task:
raise HTTPException(status_code=404, detail="Task not found")
if task.user_id != current_user.id:
raise HTTPException(status_code=403, detail="Not enough permissions")
return task
@router.patch("/{task_id}", response_model=TaskRead)
def update_task(
current_user: CurrentUserDep,
task_id: int,
task_in: TaskUpdate,
session: Session = Depends(get_session)
):
task = session.get(Task, task_id)
if not task:
raise HTTPException(status_code=404, detail="Task not found")
if task.user_id != current_user.id:
raise HTTPException(status_code=403, detail="Not enough permissions")
update_data = task_in.model_dump(exclude_unset=True)
for key, value in update_data.items():
setattr(task, key, value)
task.updated_at = datetime.utcnow()
session.add(task)
session.commit()
session.refresh(task)
return task
@router.patch("/{task_id}/toggle", response_model=TaskRead)
def toggle_task(
current_user: CurrentUserDep,
task_id: int,
session: Session = Depends(get_session)
):
task = session.get(Task, task_id)
if not task:
raise HTTPException(status_code=404, detail="Task not found")
if task.user_id != current_user.id:
raise HTTPException(status_code=403, detail="Not enough permissions")
task.completed = not task.completed
task.completed_at = datetime.utcnow() if task.completed else None
task.updated_at = datetime.utcnow()
session.add(task)
session.commit()
session.refresh(task)
return task
@router.delete("/{task_id}")
def delete_task(
current_user: CurrentUserDep,
task_id: int,
session: Session = Depends(get_session)
):
task = session.get(Task, task_id)
if not task:
raise HTTPException(status_code=404, detail="Task not found")
if task.user_id != current_user.id:
raise HTTPException(status_code=403, detail="Not enough permissions")
session.delete(task)
session.commit()
return {"message": "Task deleted successfully"}
|