79 lines
2.1 KiB
Python
79 lines
2.1 KiB
Python
"""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
|