"""DMS versioning — API views. Endpoints --------- Lock status: GET /dms/documents/{doc_id}/lock/ current active lock or {locked: false} Check-out / check-in: POST /dms/documents/{doc_id}/checkout/ acquire exclusive lock POST /dms/documents/{doc_id}/checkin/ release lock (+ optional new version) POST /dms/documents/{doc_id}/release-lock/ admin: force-release any lock History: GET /dms/documents/{doc_id}/checkout-history/ audit log of checkout events Version diff: GET /dms/documents/{doc_id}/versions/diff/ list diffs for document POST /dms/documents/{doc_id}/versions/diff/ request a new diff GET /dms/documents/{doc_id}/versions/diff/{diff_id}/ retrieve diff """ from __future__ import annotations from rest_framework import status from rest_framework.decorators import api_view, permission_classes from rest_framework.permissions import IsAuthenticated from rest_framework.request import Request from rest_framework.response import Response from simorgh.apps.dms.common.exceptions import AssetNotFound from simorgh.apps.dms.versioning import queries, services from simorgh.apps.dms.versioning.iam_permissions import ( PERM_VERSIONING_CHECKIN, PERM_VERSIONING_CHECKOUT, PERM_VERSIONING_COMPARE, PERM_VERSIONING_RELEASE, ) from simorgh.apps.dms.versioning.serializers import ( CheckinInputSerializer, CheckoutInputSerializer, CheckoutSessionSerializer, DocumentLockSerializer, ForceReleaseInputSerializer, VersionDiffRequestSerializer, VersionDiffSerializer, ) from simorgh.apps.dms.versioning.services import LockConflictError, LockError, VersioningError # --------------------------------------------------------------------------- # Shared helpers (mirrors pattern from documents/views.py) # --------------------------------------------------------------------------- def _tenant(request: Request): tenant = getattr(request, "tenant", None) if tenant is None: from rest_framework.exceptions import PermissionDenied raise PermissionDenied("Tenant required.") return tenant def _org_node_id(request: Request) -> int: from simorgh.apps.memberships.models import Membership, MembershipStatus t = _tenant(request) m = ( Membership.objects.filter( tenant=t, users=request.user, status=MembershipStatus.ACTIVE ) .order_by("id") .first() ) if m is not None: return m.organization_node_id if getattr(request.user, "is_superuser", False): from simorgh.apps.organizations.models import OrganizationNode root = OrganizationNode.objects.filter(tenant=t).order_by("id").first() if root is not None: return root.pk from rest_framework.exceptions import PermissionDenied raise PermissionDenied("No active membership for this tenant.") def _require_perm(request: Request, codename: str) -> None: from rest_framework.exceptions import PermissionDenied from simorgh.core.context import current_request_context ctx = current_request_context() if ctx.is_superuser: return if codename not in (ctx.permissions or set()): raise PermissionDenied(f"Permission required: {codename}") def _get_document(request: Request, doc_id: str): from simorgh.apps.dms.documents.queries import get_document tenant = _tenant(request) return get_document(tenant.pk, doc_id) # --------------------------------------------------------------------------- # Lock status # --------------------------------------------------------------------------- @api_view(["GET"]) @permission_classes([IsAuthenticated]) def document_lock_status(request: Request, doc_id: str) -> Response: """GET /dms/documents/{doc_id}/lock/ — return current active lock or not-locked payload.""" _require_perm(request, PERM_VERSIONING_CHECKOUT) try: doc = _get_document(request, doc_id) except AssetNotFound as exc: return Response({"error": str(exc)}, status=status.HTTP_404_NOT_FOUND) lock = queries.get_active_lock(doc) if lock is None: return Response({"locked": False, "lock": None}) return Response({"locked": True, "lock": DocumentLockSerializer(lock).data}) # --------------------------------------------------------------------------- # Check-out # --------------------------------------------------------------------------- @api_view(["POST"]) @permission_classes([IsAuthenticated]) def checkout_document(request: Request, doc_id: str) -> Response: """POST /dms/documents/{doc_id}/checkout/""" _require_perm(request, PERM_VERSIONING_CHECKOUT) try: doc = _get_document(request, doc_id) except AssetNotFound as exc: return Response({"error": str(exc)}, status=status.HTTP_404_NOT_FOUND) ser = CheckoutInputSerializer(data=request.data) ser.is_valid(raise_exception=True) d = ser.validated_data tenant = _tenant(request) org_node_id = _org_node_id(request) try: lock = services.checkout_document( doc, user=request.user, tenant_id=tenant.pk, organization_node_id=org_node_id, expires_at=d.get("expires_at"), notes=d.get("notes", ""), ) except LockConflictError as exc: return Response({"error": str(exc)}, status=status.HTTP_409_CONFLICT) except (LockError, Exception) as exc: from simorgh.apps.dms.documents.services import DocumentError if isinstance(exc, DocumentError): return Response({"error": str(exc)}, status=status.HTTP_400_BAD_REQUEST) raise return Response(DocumentLockSerializer(lock).data, status=status.HTTP_201_CREATED) # --------------------------------------------------------------------------- # Check-in # --------------------------------------------------------------------------- @api_view(["POST"]) @permission_classes([IsAuthenticated]) def checkin_document(request: Request, doc_id: str) -> Response: """POST /dms/documents/{doc_id}/checkin/""" _require_perm(request, PERM_VERSIONING_CHECKIN) try: doc = _get_document(request, doc_id) except AssetNotFound as exc: return Response({"error": str(exc)}, status=status.HTTP_404_NOT_FOUND) lock = queries.get_active_lock(doc) if lock is None: return Response( {"error": "Document is not currently checked out."}, status=status.HTTP_400_BAD_REQUEST, ) # Only the lock holder — or someone with RELEASE permission — may check in. from simorgh.core.context import current_request_context ctx = current_request_context() is_privileged = ctx.is_superuser or PERM_VERSIONING_RELEASE in (ctx.permissions or set()) if lock.locked_by_id != request.user.pk and not is_privileged: return Response( {"error": "You do not hold the checkout lock for this document."}, status=status.HTTP_403_FORBIDDEN, ) ser = CheckinInputSerializer(data=request.data) ser.is_valid(raise_exception=True) d = ser.validated_data file_asset = None if d.get("file_asset_id"): from simorgh.apps.storage.models import FileMetadata import uuid tenant = _tenant(request) try: file_asset = FileMetadata.objects.get( public_id=uuid.UUID(str(d["file_asset_id"])), tenant_id=tenant.pk, is_deleted=False, ) except FileMetadata.DoesNotExist: return Response( {"error": "File asset not found."}, status=status.HTTP_404_NOT_FOUND ) try: lock, version = services.checkin_document( lock, file_asset=file_asset, bump=d.get("bump", "minor"), label=d.get("label", ""), change_summary=d.get("change_summary", ""), lock_token=d.get("lock_token"), ) except LockError as exc: return Response({"error": str(exc)}, status=status.HTTP_400_BAD_REQUEST) data = DocumentLockSerializer(lock).data if version is not None: from simorgh.apps.dms.documents.serializers import DocumentVersionSerializer data["version_created"] = DocumentVersionSerializer(version).data else: data["version_created"] = None return Response(data, status=status.HTTP_200_OK) # --------------------------------------------------------------------------- # Force release # --------------------------------------------------------------------------- @api_view(["POST"]) @permission_classes([IsAuthenticated]) def force_release_lock(request: Request, doc_id: str) -> Response: """POST /dms/documents/{doc_id}/release-lock/""" _require_perm(request, PERM_VERSIONING_RELEASE) try: doc = _get_document(request, doc_id) except AssetNotFound as exc: return Response({"error": str(exc)}, status=status.HTTP_404_NOT_FOUND) lock = queries.get_active_lock(doc) if lock is None: return Response( {"error": "Document is not currently locked."}, status=status.HTTP_400_BAD_REQUEST, ) ser = ForceReleaseInputSerializer(data=request.data) ser.is_valid(raise_exception=True) try: lock = services.force_release_lock( lock, released_by=request.user, reason=ser.validated_data.get("reason", ""), ) except LockError as exc: return Response({"error": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(DocumentLockSerializer(lock).data, status=status.HTTP_200_OK) # --------------------------------------------------------------------------- # Checkout history # --------------------------------------------------------------------------- @api_view(["GET"]) @permission_classes([IsAuthenticated]) def checkout_history(request: Request, doc_id: str) -> Response: """GET /dms/documents/{doc_id}/checkout-history/""" _require_perm(request, PERM_VERSIONING_CHECKOUT) try: doc = _get_document(request, doc_id) except AssetNotFound as exc: return Response({"error": str(exc)}, status=status.HTTP_404_NOT_FOUND) events = queries.list_checkout_history(doc) return Response(CheckoutSessionSerializer(events, many=True).data) # --------------------------------------------------------------------------- # Version diff # --------------------------------------------------------------------------- @api_view(["GET", "POST"]) @permission_classes([IsAuthenticated]) def version_diff_list(request: Request, doc_id: str) -> Response: """ GET /dms/documents/{doc_id}/versions/diff/ — list diffs for this document POST /dms/documents/{doc_id}/versions/diff/ — request a new diff """ _require_perm(request, PERM_VERSIONING_COMPARE) try: doc = _get_document(request, doc_id) except AssetNotFound as exc: return Response({"error": str(exc)}, status=status.HTTP_404_NOT_FOUND) if request.method == "GET": diffs = queries.list_version_diffs_for_document(doc) return Response(VersionDiffSerializer(diffs, many=True).data) # POST — request diff ser = VersionDiffRequestSerializer(data=request.data) ser.is_valid(raise_exception=True) d = ser.validated_data from simorgh.apps.dms.documents.queries import get_version try: from_version = get_version(doc, str(d["from_version_id"])) to_version = get_version(doc, str(d["to_version_id"])) except AssetNotFound as exc: return Response({"error": str(exc)}, status=status.HTTP_404_NOT_FOUND) tenant = _tenant(request) org_node_id = _org_node_id(request) try: diff = services.request_version_diff( from_version, to_version, tenant_id=tenant.pk, organization_node_id=org_node_id, diff_type=d.get("diff_type", "full"), ) except VersioningError as exc: return Response({"error": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(VersionDiffSerializer(diff).data, status=status.HTTP_201_CREATED) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def version_diff_detail(request: Request, doc_id: str, diff_id: str) -> Response: """GET /dms/documents/{doc_id}/versions/diff/{diff_id}/""" _require_perm(request, PERM_VERSIONING_COMPARE) try: _get_document(request, doc_id) # validate document access except AssetNotFound as exc: return Response({"error": str(exc)}, status=status.HTTP_404_NOT_FOUND) tenant = _tenant(request) try: diff = queries.get_version_diff(tenant.pk, diff_id) except AssetNotFound as exc: return Response({"error": str(exc)}, status=status.HTTP_404_NOT_FOUND) return Response(VersionDiffSerializer(diff).data)