|
| 1 | +import uuid |
| 2 | +from typing import Any |
| 3 | + |
| 4 | +from fastapi import APIRouter, HTTPException |
| 5 | +from sqlmodel import func, select |
| 6 | + |
| 7 | +from app.api.deps import CurrentUser, SessionDep |
| 8 | +from app.models import SubTodo, SubTodoCreate, SubTodoPublic, SubTodosPublic, SubTodoUpdate, Message, Todo |
| 9 | + |
| 10 | +router = APIRouter(prefix="/subtodos", tags=["subtodos"]) |
| 11 | + |
| 12 | + |
| 13 | +@router.get("/", response_model=SubTodosPublic) |
| 14 | +def read_sub_todos( |
| 15 | + session: SessionDep, current_user: CurrentUser, todo_id=uuid.UUID, skip: int = 0, limit: int = 100 |
| 16 | +) -> Any: |
| 17 | + """ |
| 18 | + Retrieve sub todos. |
| 19 | + """ |
| 20 | + |
| 21 | + if current_user.is_superuser: |
| 22 | + count_statement = select(func.count()).select_from(SubTodo) |
| 23 | + count = session.exec(count_statement).one() |
| 24 | + statement = select(SubTodo).offset(skip).limit(limit) |
| 25 | + todos = session.exec(statement).all() |
| 26 | + else: |
| 27 | + count_statement = ( |
| 28 | + select(func.count()) |
| 29 | + .select_from(SubTodo) |
| 30 | + .where(SubTodo.todo_id == todo_id) |
| 31 | + ) |
| 32 | + count = session.exec(count_statement).one() |
| 33 | + statement = ( |
| 34 | + select(SubTodo) |
| 35 | + .where(SubTodo.todo_id == todo_id) |
| 36 | + .offset(skip) |
| 37 | + .limit(limit) |
| 38 | + ) |
| 39 | + todos = session.exec(statement).all() |
| 40 | + |
| 41 | + return SubTodosPublic(data=todos, count=count) |
| 42 | + |
| 43 | + |
| 44 | +@router.get("/{id}", response_model=SubTodoPublic) |
| 45 | +def read_item(session: SessionDep, current_user: CurrentUser, id: uuid.UUID) -> Any: |
| 46 | + """ |
| 47 | + Get sub todo by ID. |
| 48 | + """ |
| 49 | + sub_todo = session.get(SubTodo, id) |
| 50 | + if not sub_todo: |
| 51 | + raise HTTPException(status_code=404, detail="Sub todo not found") |
| 52 | + return sub_todo |
| 53 | + |
| 54 | +@router.post("/", response_model=SubTodoPublic) |
| 55 | +def create_sub_todo( |
| 56 | + *, session: SessionDep, sub_todo_in: SubTodoCreate |
| 57 | +) -> Any: |
| 58 | + """ |
| 59 | + Create new sub todo. |
| 60 | + """ |
| 61 | + todo = session.get(Todo, sub_todo_in.todo_id) |
| 62 | + if not todo: |
| 63 | + raise HTTPException(status_code=404, detail="Todo not found") |
| 64 | + sub_todo = SubTodoCreate.model_validate(sub_todo_in) |
| 65 | + create_todo = SubTodo(**dict(sub_todo)) |
| 66 | + session.add(create_todo) |
| 67 | + session.commit() |
| 68 | + session.refresh(create_todo) |
| 69 | + return create_todo |
| 70 | + |
| 71 | + |
| 72 | +# @router.put("/{id}", response_model=SubTodoPublic) |
| 73 | +# def update_sub_todo( |
| 74 | +# *, |
| 75 | +# session: SessionDep, |
| 76 | +# current_user: CurrentUser, |
| 77 | +# id: uuid.UUID, |
| 78 | +# todo_in: SubTodoUpdate, |
| 79 | +# ) -> Any: |
| 80 | +# """ |
| 81 | +# Update an item. |
| 82 | +# """ |
| 83 | +# sub_todo = session.get(SubTodo, id) |
| 84 | +# if not sub_todo: |
| 85 | +# raise HTTPException(status_code=404, detail="SubTodo not found") |
| 86 | +# if not current_user.is_superuser and (sub_todo.owner_id != current_user.id): |
| 87 | +# raise HTTPException(status_code=400, detail="Not enough permissions") |
| 88 | +# update_dict = todo_in.model_dump(exclude_unset=True) |
| 89 | +# sub_todo.sqlmodel_update(update_dict) |
| 90 | +# session.add(sub_todo) |
| 91 | +# session.commit() |
| 92 | +# session.refresh(sub_todo) |
| 93 | +# return sub_todo |
| 94 | + |
| 95 | +@router.put("/{id}", response_model=SubTodoPublic) |
| 96 | +def update_sub_todo( |
| 97 | + *, |
| 98 | + session: SessionDep, |
| 99 | + current_user: CurrentUser, |
| 100 | + id: uuid.UUID, |
| 101 | + sub_todo_in: SubTodoUpdate, |
| 102 | +) -> Any: |
| 103 | + """ |
| 104 | + Update a sub todo by ID. |
| 105 | + """ |
| 106 | + # Fetch the sub todo by ID |
| 107 | + sub_todo = session.get(SubTodo, id) |
| 108 | + if not sub_todo: |
| 109 | + raise HTTPException(status_code=404, detail="SubTodo not found") |
| 110 | + |
| 111 | + # Fetch the associated parent todo for ownership verification |
| 112 | + parent_todo = session.get(Todo, sub_todo.todo_id) |
| 113 | + if not parent_todo: |
| 114 | + raise HTTPException(status_code=404, detail="Parent Todo not found") |
| 115 | + |
| 116 | + # Permission check: Ensure the user is the owner or a superuser |
| 117 | + if not current_user.is_superuser and (parent_todo.owner_id != current_user.id): |
| 118 | + raise HTTPException(status_code=403, detail="Not enough permissions") |
| 119 | + |
| 120 | + # Update the SubTodo with the new data |
| 121 | + update_data = sub_todo_in.model_dump(exclude_unset=True) |
| 122 | + for key, value in update_data.items(): |
| 123 | + setattr(sub_todo, key, value) |
| 124 | + |
| 125 | + # Commit changes to the database |
| 126 | + session.add(sub_todo) |
| 127 | + session.commit() |
| 128 | + session.refresh(sub_todo) |
| 129 | + |
| 130 | + return sub_todo |
| 131 | + |
| 132 | +@router.delete("/{id}") |
| 133 | +def delete_sub_todo( |
| 134 | + session: SessionDep, current_user: CurrentUser, id: uuid.UUID |
| 135 | +) -> Message: |
| 136 | + """ |
| 137 | + Delete a sub todo. |
| 138 | + """ |
| 139 | + sub_todo = session.get(SubTodo, id) |
| 140 | + if not sub_todo: |
| 141 | + raise HTTPException(status_code=404, detail="SubTodo not found") |
| 142 | + todo = session.get(Todo, sub_todo.todo_id) |
| 143 | + if not todo: |
| 144 | + raise HTTPException(status_code=404, detail="Todo not found") |
| 145 | + if not current_user.is_superuser and (todo.owner_id != current_user.id): |
| 146 | + raise HTTPException(status_code=400, detail="Not enough permissions") |
| 147 | + session.delete(sub_todo) |
| 148 | + session.commit() |
| 149 | + return Message(message="SubTodo deleted successfully") |
0 commit comments