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, ) from ..activity import log_activity 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 ) await log_activity(current_user, "folder_created", "folder", folder.id) return await FolderOut.from_tortoise_orm(folder) @router.get("/{folder_id}/path", response_model=List[FolderOut]) async def get_folder_path( 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" ) path = [] current = folder while current: path.insert(0, await FolderOut.from_tortoise_orm(current)) if current.parent_id: current = await Folder.get_or_none( id=current.parent_id, owner=current_user, is_deleted=False ) else: current = None return path @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() await log_activity(current_user, "folder_updated", "folder", folder.id) 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" ) folder.is_deleted = True await folder.save() await log_activity(current_user, "folder_deleted", "folder", folder.id) 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() await log_activity(current_user, "folder_starred", "folder", folder_id) 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() await log_activity(current_user, "folder_unstarred", "folder", folder_id) 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() await log_activity(current_user, "folder_deleted", "folder", folder_id) updated_folders.append(db_folder) elif batch_operation.operation == "star": db_folder.is_starred = True await db_folder.save() await log_activity(current_user, "folder_starred", "folder", folder_id) updated_folders.append(db_folder) elif batch_operation.operation == "unstar": db_folder.is_starred = False await db_folder.save() await log_activity(current_user, "folder_unstarred", "folder", folder_id) 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 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 db_folder.parent = target_folder await db_folder.save() await log_activity(current_user, "folder_moved", "folder", folder_id) updated_folders.append(db_folder) return [await FolderOut.from_tortoise_orm(f) for f in updated_folders]