95 lines
3.2 KiB
Python
95 lines
3.2 KiB
Python
"""Opaque working-state annotation endpoints."""
|
|
|
|
import json
|
|
from datetime import UTC, datetime
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.auth.dependencies import get_current_user, verify_csrf
|
|
from app.db import get_db
|
|
from app.models.annotation_state import AnnotationState
|
|
from app.models.document import Document
|
|
from app.schemas.annotations import (
|
|
AnnotationPayload,
|
|
AnnotationStateResponse,
|
|
AnnotationStateUpdateRequest,
|
|
AnnotationStateUpdateResponse,
|
|
)
|
|
|
|
router = APIRouter(prefix="/documents", tags=["annotations"])
|
|
|
|
|
|
def _now_iso() -> str:
|
|
return datetime.now(UTC).isoformat()
|
|
|
|
|
|
def _utc(value: datetime) -> datetime:
|
|
return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)
|
|
|
|
|
|
def _document_or_404(document_id: str, db: Session) -> Document:
|
|
document = db.scalar(
|
|
select(Document).where(Document.id == document_id, Document.deleted_at.is_(None))
|
|
)
|
|
if not document:
|
|
raise HTTPException(status_code=404, detail="Document not found")
|
|
return document
|
|
|
|
|
|
def _read_data(state: AnnotationState | None) -> list[AnnotationPayload]:
|
|
if not state:
|
|
return []
|
|
try:
|
|
data = json.loads(state.data)
|
|
except json.JSONDecodeError:
|
|
return []
|
|
return data if isinstance(data, list) else []
|
|
|
|
|
|
@router.get("/{document_id}/annotations", response_model=AnnotationStateResponse)
|
|
def get_annotations(
|
|
document_id: str,
|
|
db: Session = Depends(get_db),
|
|
_user_id: int = Depends(get_current_user),
|
|
) -> AnnotationStateResponse:
|
|
_document_or_404(document_id, db)
|
|
state = db.scalar(select(AnnotationState).where(AnnotationState.document_id == document_id))
|
|
updated_at = datetime.now(UTC) if not state else _utc(datetime.fromisoformat(state.updated_at))
|
|
return AnnotationStateResponse(data=_read_data(state), updatedAt=updated_at)
|
|
|
|
|
|
@router.put(
|
|
"/{document_id}/annotations",
|
|
response_model=AnnotationStateUpdateResponse,
|
|
dependencies=[Depends(verify_csrf)],
|
|
)
|
|
def update_annotations(
|
|
document_id: str,
|
|
request: AnnotationStateUpdateRequest,
|
|
db: Session = Depends(get_db),
|
|
_user_id: int = Depends(get_current_user),
|
|
) -> AnnotationStateUpdateResponse:
|
|
document = _document_or_404(document_id, db)
|
|
state = db.scalar(select(AnnotationState).where(AnnotationState.document_id == document_id))
|
|
|
|
if state and request.baseUpdatedAt:
|
|
current_updated_at = _utc(datetime.fromisoformat(state.updated_at))
|
|
request_updated_at = _utc(request.baseUpdatedAt)
|
|
if abs((current_updated_at - request_updated_at).total_seconds()) > 1.0:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail=f"Conflict: annotations have been modified since {request.baseUpdatedAt}",
|
|
)
|
|
|
|
now = _now_iso()
|
|
data_json = json.dumps(request.data, separators=(",", ":"))
|
|
if state:
|
|
state.data = data_json
|
|
state.updated_at = now
|
|
else:
|
|
db.add(AnnotationState(document_id=document_id, data=data_json, updated_at=now))
|
|
document.updated_at = now
|
|
db.commit()
|
|
return AnnotationStateUpdateResponse(updatedAt=datetime.fromisoformat(now))
|