376 lines
11 KiB
Python
376 lines
11 KiB
Python
"""Folder and document-folder organization endpoints."""
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import select, text
|
|
from sqlalchemy.exc import IntegrityError, OperationalError
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from db.models import Document, Folder, User
|
|
from deps.auth import get_regular_user
|
|
from deps.db import get_db
|
|
from deps.utils import get_client_ip
|
|
from services.audit import write_audit_log
|
|
from storage import get_storage_backend
|
|
|
|
router = APIRouter(prefix="/api/folders", tags=["folders"])
|
|
|
|
|
|
class FolderCreate(BaseModel):
|
|
name: str
|
|
parent_id: Optional[str] = None
|
|
|
|
|
|
class FolderRename(BaseModel):
|
|
name: str
|
|
|
|
|
|
class DocumentMove(BaseModel):
|
|
folder_id: Optional[str] = None
|
|
|
|
|
|
def _folder_to_dict(folder: Folder) -> dict:
|
|
return {
|
|
"id": str(folder.id),
|
|
"name": folder.name,
|
|
"parent_id": str(folder.parent_id) if folder.parent_id else None,
|
|
"user_id": str(folder.user_id),
|
|
"created_at": folder.created_at.isoformat() if folder.created_at else None,
|
|
}
|
|
|
|
|
|
def _doc_to_dict(doc: Document) -> dict:
|
|
return {
|
|
"id": str(doc.id),
|
|
"filename": doc.filename,
|
|
"content_type": doc.content_type,
|
|
"size_bytes": doc.size_bytes,
|
|
"status": doc.status,
|
|
"folder_id": str(doc.folder_id) if doc.folder_id else None,
|
|
"user_id": str(doc.user_id) if doc.user_id else None,
|
|
"created_at": doc.created_at.isoformat() if doc.created_at else None,
|
|
"updated_at": doc.updated_at.isoformat() if doc.updated_at else None,
|
|
}
|
|
|
|
|
|
def _parse_uuid(value: str, not_found_detail: str) -> uuid.UUID:
|
|
try:
|
|
return uuid.UUID(value)
|
|
except ValueError:
|
|
raise HTTPException(status_code=404, detail=not_found_detail)
|
|
|
|
|
|
async def _get_owned_folder(
|
|
session: AsyncSession,
|
|
folder_id: str | uuid.UUID,
|
|
user_id: uuid.UUID,
|
|
detail: str = "Folder not found",
|
|
) -> Folder:
|
|
uid = folder_id if isinstance(folder_id, uuid.UUID) else _parse_uuid(folder_id, detail)
|
|
folder = await session.get(Folder, uid)
|
|
if folder is None or folder.user_id != user_id:
|
|
raise HTTPException(status_code=404, detail=detail)
|
|
return folder
|
|
|
|
|
|
async def _get_owned_document(
|
|
session: AsyncSession,
|
|
doc_id: str,
|
|
user_id: uuid.UUID,
|
|
) -> Document:
|
|
uid = _parse_uuid(doc_id, "Document not found")
|
|
doc = await session.get(Document, uid)
|
|
if doc is None or doc.user_id != user_id:
|
|
raise HTTPException(status_code=404, detail="Document not found")
|
|
return doc
|
|
|
|
|
|
async def _ensure_unique_folder_name(
|
|
session: AsyncSession,
|
|
user_id: uuid.UUID,
|
|
name: str,
|
|
parent_id: Optional[uuid.UUID],
|
|
exclude_id: Optional[uuid.UUID] = None,
|
|
) -> None:
|
|
stmt = select(Folder).where(
|
|
Folder.user_id == user_id,
|
|
Folder.name == name,
|
|
Folder.parent_id == parent_id,
|
|
)
|
|
if exclude_id is not None:
|
|
stmt = stmt.where(Folder.id != exclude_id)
|
|
|
|
dup = await session.execute(stmt)
|
|
if dup.scalar_one_or_none() is not None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="A folder with that name already exists here",
|
|
)
|
|
|
|
|
|
def _duplicate_folder_error() -> HTTPException:
|
|
return HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="A folder with that name already exists here",
|
|
)
|
|
|
|
|
|
@router.post("", status_code=status.HTTP_201_CREATED)
|
|
async def create_folder(
|
|
body: FolderCreate,
|
|
request: Request,
|
|
session: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_regular_user),
|
|
):
|
|
parent_uuid: Optional[uuid.UUID] = None
|
|
if body.parent_id is not None:
|
|
parent = await _get_owned_folder(
|
|
session, body.parent_id, current_user.id, "Parent folder not found"
|
|
)
|
|
parent_uuid = parent.id
|
|
|
|
await _ensure_unique_folder_name(session, current_user.id, body.name, parent_uuid)
|
|
|
|
folder = Folder(
|
|
user_id=current_user.id,
|
|
name=body.name,
|
|
parent_id=parent_uuid,
|
|
)
|
|
try:
|
|
session.add(folder)
|
|
await session.commit()
|
|
except IntegrityError:
|
|
await session.rollback()
|
|
raise _duplicate_folder_error()
|
|
|
|
await write_audit_log(
|
|
session,
|
|
event_type="folder.created",
|
|
user_id=current_user.id,
|
|
actor_id=current_user.id,
|
|
resource_id=folder.id,
|
|
ip_address=get_client_ip(request),
|
|
metadata_={"name": folder.name, "parent_id": str(parent_uuid) if parent_uuid else None},
|
|
)
|
|
|
|
return _folder_to_dict(folder)
|
|
|
|
|
|
@router.get("")
|
|
async def list_folders(
|
|
parent_id: Optional[str] = None,
|
|
session: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_regular_user),
|
|
):
|
|
parent_uuid: Optional[uuid.UUID] = None
|
|
if parent_id is not None:
|
|
parent_folder = await _get_owned_folder(
|
|
session, parent_id, current_user.id, "Parent folder not found"
|
|
)
|
|
parent_uuid = parent_folder.id
|
|
|
|
where_clause = Folder.parent_id.is_(None) if parent_uuid is None else Folder.parent_id == parent_uuid
|
|
|
|
result = await session.execute(
|
|
select(Folder)
|
|
.where(Folder.user_id == current_user.id, where_clause)
|
|
.order_by(Folder.name)
|
|
)
|
|
folders = result.scalars().all()
|
|
|
|
folder_ids = [f.id for f in folders]
|
|
folders_with_children: set = set()
|
|
if folder_ids:
|
|
children_result = await session.execute(
|
|
select(Folder.parent_id.distinct()).where(
|
|
Folder.user_id == current_user.id,
|
|
Folder.parent_id.in_(folder_ids),
|
|
)
|
|
)
|
|
folders_with_children = {row[0] for row in children_result}
|
|
|
|
return {
|
|
"items": [
|
|
{**_folder_to_dict(f), "has_children": f.id in folders_with_children}
|
|
for f in folders
|
|
]
|
|
}
|
|
|
|
|
|
@router.get("/{folder_id}")
|
|
async def get_folder(
|
|
folder_id: str,
|
|
session: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_regular_user),
|
|
):
|
|
folder = await _get_owned_folder(session, folder_id, current_user.id)
|
|
|
|
crumbs = [{"id": str(folder.id), "name": folder.name}]
|
|
current = folder
|
|
visited: set = {current.id}
|
|
while current.parent_id is not None:
|
|
if current.parent_id in visited:
|
|
break
|
|
parent = await session.get(Folder, current.parent_id)
|
|
if parent is None or parent.user_id != current_user.id:
|
|
break
|
|
visited.add(parent.id)
|
|
crumbs.append({"id": str(parent.id), "name": parent.name})
|
|
current = parent
|
|
|
|
crumbs.reverse()
|
|
|
|
response = _folder_to_dict(folder)
|
|
response["breadcrumb"] = crumbs
|
|
return response
|
|
|
|
|
|
@router.patch("/{folder_id}")
|
|
async def rename_folder(
|
|
folder_id: str,
|
|
body: FolderRename,
|
|
request: Request,
|
|
session: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_regular_user),
|
|
):
|
|
folder = await _get_owned_folder(session, folder_id, current_user.id)
|
|
old_name = folder.name
|
|
|
|
if body.name != folder.name:
|
|
await _ensure_unique_folder_name(
|
|
session, current_user.id, body.name, folder.parent_id, folder.id
|
|
)
|
|
|
|
folder.name = body.name
|
|
try:
|
|
await session.commit()
|
|
except IntegrityError:
|
|
await session.rollback()
|
|
raise _duplicate_folder_error()
|
|
|
|
await write_audit_log(
|
|
session,
|
|
event_type="folder.renamed",
|
|
user_id=current_user.id,
|
|
actor_id=current_user.id,
|
|
resource_id=folder.id,
|
|
ip_address=get_client_ip(request),
|
|
metadata_={"old_name": old_name, "new_name": folder.name},
|
|
)
|
|
|
|
return _folder_to_dict(folder)
|
|
|
|
|
|
@router.delete("/{folder_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def delete_folder(
|
|
folder_id: str,
|
|
request: Request,
|
|
session: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_regular_user),
|
|
):
|
|
folder = await _get_owned_folder(session, folder_id, current_user.id)
|
|
folder_name = folder.name
|
|
|
|
subtree_folder_ids: list[str] = []
|
|
try:
|
|
cte_result = await session.execute(
|
|
text(
|
|
"WITH RECURSIVE subtree AS ("
|
|
" SELECT id FROM folders WHERE id = :root_id AND user_id = :uid "
|
|
" UNION ALL "
|
|
" SELECT f.id FROM folders f "
|
|
" JOIN subtree s ON f.parent_id = s.id "
|
|
" WHERE f.user_id = :uid"
|
|
") SELECT id FROM subtree"
|
|
),
|
|
{"root_id": folder.id.hex, "uid": current_user.id.hex},
|
|
)
|
|
subtree_folder_ids = [str(row[0]) for row in cte_result.fetchall()]
|
|
except OperationalError:
|
|
subtree_folder_ids = [str(folder.id)]
|
|
|
|
if subtree_folder_ids:
|
|
subtree_uuids = []
|
|
for fid in subtree_folder_ids:
|
|
try:
|
|
subtree_uuids.append(uuid.UUID(fid))
|
|
except ValueError:
|
|
pass
|
|
|
|
if subtree_uuids:
|
|
docs_result = await session.execute(
|
|
select(Document).where(
|
|
Document.folder_id.in_(subtree_uuids),
|
|
Document.user_id == current_user.id,
|
|
)
|
|
)
|
|
docs = docs_result.scalars().all()
|
|
else:
|
|
docs = []
|
|
else:
|
|
docs = []
|
|
|
|
total_bytes = sum(d.size_bytes for d in docs)
|
|
|
|
if total_bytes > 0:
|
|
await session.execute(
|
|
text(
|
|
"UPDATE quotas "
|
|
"SET used_bytes = "
|
|
"CASE WHEN used_bytes > :delta THEN used_bytes - :delta ELSE 0 END "
|
|
"WHERE user_id = :uid"
|
|
),
|
|
{"delta": total_bytes, "uid": current_user.id.hex},
|
|
)
|
|
|
|
storage_backend = get_storage_backend()
|
|
for doc in docs:
|
|
try:
|
|
await storage_backend.delete_object(doc.object_key)
|
|
except Exception:
|
|
pass
|
|
|
|
for doc in docs:
|
|
await session.delete(doc)
|
|
|
|
await session.delete(folder)
|
|
await session.commit()
|
|
|
|
await write_audit_log(
|
|
session,
|
|
event_type="folder.deleted",
|
|
user_id=current_user.id,
|
|
actor_id=current_user.id,
|
|
resource_id=folder.id,
|
|
ip_address=get_client_ip(request),
|
|
metadata_={"name": folder_name, "doc_count": len(docs)},
|
|
)
|
|
|
|
|
|
document_move_router = APIRouter(prefix="/api/documents", tags=["folders"])
|
|
|
|
|
|
@document_move_router.patch("/{doc_id}/folder")
|
|
async def move_document(
|
|
doc_id: str,
|
|
body: DocumentMove,
|
|
request: Request,
|
|
session: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_regular_user),
|
|
):
|
|
doc = await _get_owned_document(session, doc_id, current_user.id)
|
|
|
|
target_folder_uuid: Optional[uuid.UUID] = None
|
|
if body.folder_id is not None:
|
|
target_folder = await _get_owned_folder(session, body.folder_id, current_user.id)
|
|
target_folder_uuid = target_folder.id
|
|
|
|
doc.folder_id = target_folder_uuid
|
|
await session.commit()
|
|
|
|
return _doc_to_dict(doc)
|