"""DMS repositories — API views. Endpoints --------- Repositories: GET /dms/repositories/ list repositories POST /dms/repositories/ create repository GET /dms/repositories/{repo_id}/ retrieve repository PATCH /dms/repositories/{repo_id}/ update repository DELETE /dms/repositories/{repo_id}/ soft-delete repository Folders (nested under a repository): GET /dms/repositories/{repo_id}/folders/ list root folders POST /dms/repositories/{repo_id}/folders/ create root folder GET /dms/repositories/{repo_id}/folders/{folder_id}/ retrieve folder PATCH /dms/repositories/{repo_id}/folders/{folder_id}/ rename folder DELETE /dms/repositories/{repo_id}/folders/{folder_id}/ soft-delete folder GET /dms/repositories/{repo_id}/folders/{folder_id}/children/ list direct children POST /dms/repositories/{repo_id}/folders/{folder_id}/move/ move folder """ 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.repositories import queries, services from simorgh.apps.dms.repositories.permissions import ( PERM_FOLDER_MANAGE, PERM_FOLDER_VIEW, PERM_REPOSITORY_MANAGE, PERM_REPOSITORY_VIEW, ) from simorgh.apps.dms.repositories.serializers import ( FolderCreateInputSerializer, FolderMoveInputSerializer, FolderRenameInputSerializer, FolderSerializer, RepositoryCreateInputSerializer, RepositorySerializer, RepositoryUpdateInputSerializer, ) from simorgh.apps.dms.repositories.services import FolderError, RepositoryError # --------------------------------------------------------------------------- # Helpers (shared with assets views — duplicated intentionally to avoid # a cross-module import cycle; could be moved to a shared module later) # --------------------------------------------------------------------------- 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 m = ( Membership.objects.filter( tenant=_tenant(request), 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=_tenant(request)) .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}") # --------------------------------------------------------------------------- # Repository list / create # --------------------------------------------------------------------------- @api_view(["GET", "POST"]) @permission_classes([IsAuthenticated]) def repository_list(request: Request) -> Response: """ GET — list repositories for the current tenant. POST — create a new repository. """ tenant = _tenant(request) if request.method == "GET": _require_perm(request, PERM_REPOSITORY_VIEW) qs = queries.list_repositories( tenant_id=tenant.pk, workspace_id=None, search=request.query_params.get("search"), ) return Response(RepositorySerializer(qs, many=True).data) # POST _require_perm(request, PERM_REPOSITORY_MANAGE) ser = RepositoryCreateInputSerializer(data=request.data) ser.is_valid(raise_exception=True) d = ser.validated_data workspace_id = None raw_ws = d.get("workspace_id") if raw_ws is not None: from simorgh.apps.workspaces.models import Workspace try: workspace_id = Workspace.objects.get( tenant=tenant, public_id=raw_ws ).pk except Workspace.DoesNotExist: return Response( {"detail": "Workspace not found."}, status=status.HTTP_400_BAD_REQUEST ) try: repo = services.create_repository( name=d["name"], slug=d["slug"], tenant_id=tenant.pk, organization_node_id=_org_node_id(request), description=d.get("description", ""), workspace_id=workspace_id, is_default=d.get("is_default", False), max_size_bytes=d.get("max_size_bytes"), ) except RepositoryError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(RepositorySerializer(repo).data, status=status.HTTP_201_CREATED) # --------------------------------------------------------------------------- # Repository detail / update / delete # --------------------------------------------------------------------------- @api_view(["GET", "PATCH", "DELETE"]) @permission_classes([IsAuthenticated]) def repository_detail(request: Request, repo_id: str) -> Response: """ GET — retrieve repository detail. PATCH — update repository (name, description, max_size_bytes). DELETE — soft-delete repository. """ tenant = _tenant(request) try: repo = queries.get_repository(tenant.pk, repo_id) except AssetNotFound as exc: return Response({"detail": str(exc)}, status=status.HTTP_404_NOT_FOUND) if request.method == "GET": _require_perm(request, PERM_REPOSITORY_VIEW) return Response(RepositorySerializer(repo).data) if request.method == "PATCH": _require_perm(request, PERM_REPOSITORY_MANAGE) ser = RepositoryUpdateInputSerializer(data=request.data, partial=True) ser.is_valid(raise_exception=True) d = ser.validated_data services.update_repository( repo, name=d.get("name"), description=d.get("description"), max_size_bytes=d.get("max_size_bytes", ...), # type: ignore[arg-type] ) return Response(RepositorySerializer(repo).data) # DELETE _require_perm(request, PERM_REPOSITORY_MANAGE) services.delete_repository(repo) return Response(status=status.HTTP_204_NO_CONTENT) # --------------------------------------------------------------------------- # Folder list (root) / create # --------------------------------------------------------------------------- @api_view(["GET", "POST"]) @permission_classes([IsAuthenticated]) def folder_list(request: Request, repo_id: str) -> Response: """ GET — list root folders in the repository. POST — create a root-level folder. """ tenant = _tenant(request) try: repo = queries.get_repository(tenant.pk, repo_id) except AssetNotFound as exc: return Response({"detail": str(exc)}, status=status.HTTP_404_NOT_FOUND) if request.method == "GET": _require_perm(request, PERM_FOLDER_VIEW) from django.core.cache import cache from simorgh.apps.dms.cache import folder_tree_key, FOLDER_TREE_TTL cache_key = folder_tree_key(tenant.pk, str(repo.public_id)) cached = cache.get(cache_key) if cached is not None: return Response(cached) data = FolderSerializer(queries.list_root_folders(repo), many=True).data cache.set(cache_key, data, timeout=FOLDER_TREE_TTL) return Response(data) # POST _require_perm(request, PERM_FOLDER_MANAGE) ser = FolderCreateInputSerializer(data=request.data) ser.is_valid(raise_exception=True) d = ser.validated_data try: folder = services.create_folder( repository=repo, name=d["name"], parent=None, tenant_id=tenant.pk, organization_node_id=_org_node_id(request), sort_order=d.get("sort_order", 0), color=d.get("color", ""), icon=d.get("icon", ""), ) except FolderError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) from simorgh.apps.dms.cache import invalidate_repo_folders invalidate_repo_folders(tenant.pk, str(repo.public_id)) return Response(FolderSerializer(folder).data, status=status.HTTP_201_CREATED) # --------------------------------------------------------------------------- # Folder detail / rename / delete # --------------------------------------------------------------------------- @api_view(["GET", "PATCH", "DELETE"]) @permission_classes([IsAuthenticated]) def folder_detail(request: Request, repo_id: str, folder_id: str) -> Response: """ GET — retrieve folder detail. PATCH — rename folder (pass ``name``). DELETE — soft-delete folder and all descendants. """ tenant = _tenant(request) try: repo = queries.get_repository(tenant.pk, repo_id) except AssetNotFound as exc: return Response({"detail": str(exc)}, status=status.HTTP_404_NOT_FOUND) try: folder = queries.get_folder(repo, folder_id) except AssetNotFound as exc: return Response({"detail": str(exc)}, status=status.HTTP_404_NOT_FOUND) if request.method == "GET": _require_perm(request, PERM_FOLDER_VIEW) return Response(FolderSerializer(folder).data) if request.method == "PATCH": _require_perm(request, PERM_FOLDER_MANAGE) ser = FolderRenameInputSerializer(data=request.data) ser.is_valid(raise_exception=True) try: folder = services.rename_folder(folder, new_name=ser.validated_data["name"]) except FolderError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) from simorgh.apps.dms.cache import invalidate_repo_folders invalidate_repo_folders(tenant.pk, str(repo.public_id)) return Response(FolderSerializer(folder).data) # DELETE _require_perm(request, PERM_FOLDER_MANAGE) services.delete_folder(folder) from simorgh.apps.dms.cache import invalidate_repo_folders invalidate_repo_folders(tenant.pk, str(repo.public_id)) return Response(status=status.HTTP_204_NO_CONTENT) # --------------------------------------------------------------------------- # Folder children # --------------------------------------------------------------------------- @api_view(["GET"]) @permission_classes([IsAuthenticated]) def folder_children(request: Request, repo_id: str, folder_id: str) -> Response: """GET — list direct children of a folder.""" tenant = _tenant(request) try: repo = queries.get_repository(tenant.pk, repo_id) except AssetNotFound as exc: return Response({"detail": str(exc)}, status=status.HTTP_404_NOT_FOUND) try: folder = queries.get_folder(repo, folder_id) except AssetNotFound as exc: return Response({"detail": str(exc)}, status=status.HTTP_404_NOT_FOUND) _require_perm(request, PERM_FOLDER_VIEW) from django.core.cache import cache from simorgh.apps.dms.cache import folder_children_key, FOLDER_TREE_TTL cache_key = folder_children_key(tenant.pk, str(folder.public_id)) cached = cache.get(cache_key) if cached is not None: return Response(cached) data = FolderSerializer(queries.list_children(folder), many=True).data cache.set(cache_key, data, timeout=FOLDER_TREE_TTL) return Response(data) # --------------------------------------------------------------------------- # Move folder # --------------------------------------------------------------------------- @api_view(["POST"]) @permission_classes([IsAuthenticated]) def folder_move(request: Request, repo_id: str, folder_id: str) -> Response: """POST — move a folder to a new parent (or to root if parent_id is null).""" tenant = _tenant(request) try: repo = queries.get_repository(tenant.pk, repo_id) except AssetNotFound as exc: return Response({"detail": str(exc)}, status=status.HTTP_404_NOT_FOUND) try: folder = queries.get_folder(repo, folder_id) except AssetNotFound as exc: return Response({"detail": str(exc)}, status=status.HTTP_404_NOT_FOUND) _require_perm(request, PERM_FOLDER_MANAGE) ser = FolderMoveInputSerializer(data=request.data) ser.is_valid(raise_exception=True) raw_parent_id = ser.validated_data.get("parent_id") new_parent = None if raw_parent_id is not None: try: new_parent = queries.get_folder(repo, str(raw_parent_id)) except AssetNotFound as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) try: folder = services.move_folder(folder, new_parent=new_parent) except FolderError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) from simorgh.apps.dms.cache import invalidate_repo_folders invalidate_repo_folders(tenant.pk, str(repo.public_id)) return Response(FolderSerializer(folder).data)