Created ability for Admins to review deleted tasks and restore them if needed

This commit is contained in:
NPS Agent
2026-05-12 09:53:51 +09:30
parent 60a1cf1b67
commit 62cfeb0da4
8 changed files with 115 additions and 24 deletions
+24 -3
View File
@@ -94,7 +94,13 @@ def delete_user(user_id: str, db: Session = Depends(get_db), current_user: model
@app.get("/tasks", response_model=List[schemas.Task])
def read_tasks(db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
return db.query(models.Task).all()
return db.query(models.Task).filter(models.Task.deleted_at == None).all()
@app.get("/tasks/deleted", response_model=List[schemas.Task])
def read_deleted_tasks(db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
if current_user.account_type != "admin":
raise HTTPException(status_code=403, detail="Not enough permissions")
return db.query(models.Task).filter(models.Task.deleted_at != None).all()
@app.post("/tasks", response_model=schemas.Task)
def create_task(task: schemas.TaskCreate, db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
@@ -144,9 +150,24 @@ def delete_task(task_id: str, db: Session = Depends(get_db), current_user: model
if not db_task:
raise HTTPException(status_code=404, detail="Task not found")
db.delete(db_task)
from sqlalchemy.sql import func
db_task.deleted_at = func.now()
db.commit()
return {"message": "Task deleted"}
return {"message": "Task moved to trash"}
@app.post("/tasks/{task_id}/restore", response_model=schemas.Task)
def restore_task(task_id: str, db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
if current_user.account_type != "admin":
raise HTTPException(status_code=403, detail="Not enough permissions")
db_task = db.query(models.Task).filter(models.Task.id == task_id).first()
if not db_task:
raise HTTPException(status_code=404, detail="Task not found")
db_task.deleted_at = None
db.commit()
db.refresh(db_task)
return db_task
@app.get("/workspace", response_model=schemas.Workspace)
def read_workspace(db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
ws = db.query(models.Workspace).first()