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]
Reference in New Issue View Git Blame Copy Permalink