"""Shared constants, request models, and access helpers for document routes.""" from __future__ import annotations import uuid from typing import Optional from fastapi import HTTPException from pydantic import BaseModel, Field, field_validator from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from db.models import Document, Share _CLOUD_PROVIDERS = frozenset({"google_drive", "onedrive", "nextcloud", "webdav"}) class UploadUrlRequest(BaseModel): filename: str content_type: str class DocumentPatch(BaseModel): filename: Optional[str] = Field(None, min_length=1, max_length=255) folder_id: Optional[uuid.UUID] = None @field_validator("filename") @classmethod def filename_no_path_separators(cls, v: Optional[str]) -> Optional[str]: if v is not None and ("/" in v or "\\" in v): raise ValueError("filename must not contain path separators") return v def _document_not_found() -> HTTPException: return HTTPException(status_code=404, detail="Document not found") def parse_document_uuid(doc_id: str) -> uuid.UUID: try: return uuid.UUID(doc_id) except ValueError: raise _document_not_found() async def get_owned_document( session: AsyncSession, doc_id: str, user_id: uuid.UUID, ) -> Document: uid = parse_document_uuid(doc_id) doc = await session.get(Document, uid) if doc is None or doc.user_id != user_id: raise _document_not_found() return doc async def get_accessible_document( session: AsyncSession, doc_id: str, user_id: uuid.UUID, ) -> tuple[Document, bool]: uid = parse_document_uuid(doc_id) doc = await session.get(Document, uid) if doc is None: raise _document_not_found() if doc.user_id == user_id: return doc, False result = await session.execute( select(Share).where( Share.document_id == doc.id, Share.recipient_id == user_id, ) ) if result.scalar_one_or_none() is None: raise _document_not_found() return doc, True