106 lines
4.7 KiB
Python
106 lines
4.7 KiB
Python
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
||
|
|
from typing import List, Optional
|
||
|
|
|
||
|
|
from ..auth import get_current_user
|
||
|
|
from ..models import User, Folder
|
||
|
|
from ..schemas import FolderCreate, FolderOut, FolderUpdate
|
||
|
|
|
||
|
|
router = APIRouter(
|
||
|
|
prefix="/folders",
|
||
|
|
tags=["folders"],
|
||
|
|
)
|
||
|
|
|
||
|
|
@router.post("/", response_model=FolderOut, status_code=status.HTTP_201_CREATED)
|
||
|
|
async def create_folder(folder_in: FolderCreate, current_user: User = Depends(get_current_user)):
|
||
|
|
# Check if parent folder exists and belongs to the current user
|
||
|
|
parent_folder = None
|
||
|
|
if folder_in.parent_id:
|
||
|
|
parent_folder = await Folder.get_or_none(id=folder_in.parent_id, owner=current_user)
|
||
|
|
if not parent_folder:
|
||
|
|
raise HTTPException(
|
||
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
||
|
|
detail="Parent folder not found or does not belong to the current user",
|
||
|
|
)
|
||
|
|
|
||
|
|
# Check for duplicate folder name in the same parent
|
||
|
|
existing_folder = await Folder.get_or_none(
|
||
|
|
name=folder_in.name, parent=parent_folder, owner=current_user
|
||
|
|
)
|
||
|
|
if existing_folder:
|
||
|
|
raise HTTPException(
|
||
|
|
status_code=status.HTTP_409_CONFLICT,
|
||
|
|
detail="Folder with this name already exists in the current parent folder",
|
||
|
|
)
|
||
|
|
|
||
|
|
folder = await Folder.create(
|
||
|
|
name=folder_in.name, parent=parent_folder, owner=current_user
|
||
|
|
)
|
||
|
|
return await FolderOut.from_tortoise_orm(folder)
|
||
|
|
|
||
|
|
@router.get("/{folder_id}", response_model=FolderOut)
|
||
|
|
async def get_folder(folder_id: int, current_user: User = Depends(get_current_user)):
|
||
|
|
folder = await Folder.get_or_none(id=folder_id, owner=current_user, is_deleted=False)
|
||
|
|
if not folder:
|
||
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Folder not found")
|
||
|
|
return await FolderOut.from_tortoise_orm(folder)
|
||
|
|
|
||
|
|
@router.get("/", response_model=List[FolderOut])
|
||
|
|
async def list_folders(parent_id: Optional[int] = None, current_user: User = Depends(get_current_user)):
|
||
|
|
if parent_id:
|
||
|
|
parent_folder = await Folder.get_or_none(id=parent_id, owner=current_user, is_deleted=False)
|
||
|
|
if not parent_folder:
|
||
|
|
raise HTTPException(
|
||
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
||
|
|
detail="Parent folder not found or does not belong to the current user",
|
||
|
|
)
|
||
|
|
folders = await Folder.filter(parent=parent_folder, owner=current_user, is_deleted=False).order_by("name")
|
||
|
|
else:
|
||
|
|
# List root folders (folders with no parent)
|
||
|
|
folders = await Folder.filter(parent=None, owner=current_user, is_deleted=False).order_by("name")
|
||
|
|
return [await FolderOut.from_tortoise_orm(folder) for folder in folders]
|
||
|
|
|
||
|
|
@router.put("/{folder_id}", response_model=FolderOut)
|
||
|
|
async def update_folder(folder_id: int, folder_in: FolderUpdate, current_user: User = Depends(get_current_user)):
|
||
|
|
folder = await Folder.get_or_none(id=folder_id, owner=current_user, is_deleted=False)
|
||
|
|
if not folder:
|
||
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Folder not found")
|
||
|
|
|
||
|
|
if folder_in.name:
|
||
|
|
existing_folder = await Folder.get_or_none(
|
||
|
|
name=folder_in.name, parent_id=folder.parent_id, owner=current_user
|
||
|
|
)
|
||
|
|
if existing_folder and existing_folder.id != folder_id:
|
||
|
|
raise HTTPException(
|
||
|
|
status_code=status.HTTP_409_CONFLICT,
|
||
|
|
detail="Folder with this name already exists in the current parent folder",
|
||
|
|
)
|
||
|
|
folder.name = folder_in.name
|
||
|
|
|
||
|
|
if folder_in.parent_id is not None:
|
||
|
|
if folder_in.parent_id == folder_id:
|
||
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Cannot set folder as its own parent")
|
||
|
|
|
||
|
|
new_parent_folder = None
|
||
|
|
if folder_in.parent_id != 0: # 0 could represent moving to root
|
||
|
|
new_parent_folder = await Folder.get_or_none(id=folder_in.parent_id, owner=current_user, is_deleted=False)
|
||
|
|
if not new_parent_folder:
|
||
|
|
raise HTTPException(
|
||
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
||
|
|
detail="New parent folder not found or does not belong to the current user",
|
||
|
|
)
|
||
|
|
folder.parent = new_parent_folder
|
||
|
|
|
||
|
|
await folder.save()
|
||
|
|
return await FolderOut.from_tortoise_orm(folder)
|
||
|
|
|
||
|
|
@router.delete("/{folder_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||
|
|
async def delete_folder(folder_id: int, current_user: User = Depends(get_current_user)):
|
||
|
|
folder = await Folder.get_or_none(id=folder_id, owner=current_user, is_deleted=False)
|
||
|
|
if not folder:
|
||
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Folder not found")
|
||
|
|
|
||
|
|
# Soft delete
|
||
|
|
folder.is_deleted = True
|
||
|
|
await folder.save()
|
||
|
|
return
|