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, BatchFolderOperation, BatchMoveCopyPayload
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
@router.post("/{folder_id}/star", response_model=FolderOut)
async def star_folder(folder_id: int, current_user: User = Depends(get_current_user)):
db_folder = await Folder.get_or_none(id=folder_id, owner=current_user, is_deleted=False)
if not db_folder:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Folder not found")
db_folder.is_starred = True
await db_folder.save()
return await FolderOut.from_tortoise_orm(db_folder)
@router.post("/{folder_id}/unstar", response_model=FolderOut)
async def unstar_folder(folder_id: int, current_user: User = Depends(get_current_user)):
db_folder = await Folder.get_or_none(id=folder_id, owner=current_user, is_deleted=False)
if not db_folder:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Folder not found")
db_folder.is_starred = False
await db_folder.save()
return await FolderOut.from_tortoise_orm(db_folder)
@router.post("/batch", response_model=List[FolderOut])
async def batch_folder_operations(
batch_operation: BatchFolderOperation,
payload: Optional[BatchMoveCopyPayload] = None,
current_user: User = Depends(get_current_user)
):
updated_folders = []
for folder_id in batch_operation.folder_ids:
db_folder = await Folder.get_or_none(id=folder_id, owner=current_user, is_deleted=False)
if not db_folder:
continue # Skip if folder not found or not owned by user
if batch_operation.operation == "delete":
db_folder.is_deleted = True
await db_folder.save()
updated_folders.append(db_folder)
elif batch_operation.operation == "star":
db_folder.is_starred = True
await db_folder.save()
updated_folders.append(db_folder)
elif batch_operation.operation == "unstar":
db_folder.is_starred = False
await db_folder.save()
updated_folders.append(db_folder)
elif batch_operation.operation == "move" and payload and payload.target_folder_id is not None:
target_folder = await Folder.get_or_none(id=payload.target_folder_id, owner=current_user, is_deleted=False)
if not target_folder:
continue # Skip if target folder not found
existing_folder = await Folder.get_or_none(
name=db_folder.name, parent=target_folder, owner=current_user, is_deleted=False
)
if existing_folder and existing_folder.id != folder_id:
continue # Skip if folder with same name exists
db_folder.parent = target_folder
await db_folder.save()
updated_folders.append(db_folder)
return [await FolderOut.from_tortoise_orm(f) for f in updated_folders]
Reference in New Issue View Git Blame Copy Permalink