refactor(BA-4092): Refactor VFolder purge API to Action-Service-Repository pattern

- Add PurgeVFolderAction and PurgeVFolderActionResult classes
- Add purge() method to VFolderService with status validation
- Add purge_vfolder() method to VfolderRepository using execute_purger
- Register purge_vfolder processor in VFolderProcessors
- Update API handler to use processor instead of direct DB access

Author: fregataa

changes/8317.misc.md

@@ -0,0 +1 @@
+Refactor VFolder purge API to use Action-Service-Repository pattern.

src/ai/backend/manager/api/vfolder.py

@@ -16,7 +16,6 @@
     Any,
     Concatenate,
     ParamSpec,
-    cast,
 )
 
 import aiohttp
@@ -104,6 +103,7 @@
     vfolder_status_map,
     vfolders,
 )
+from ai.backend.manager.repositories.base.purger import Purger
 from ai.backend.manager.repositories.base.updater import Updater
 from ai.backend.manager.repositories.vfolder.updaters import VFolderAttributeUpdaterSpec
 from ai.backend.manager.services.vfolder.actions.base import (
@@ -114,6 +114,7 @@
     GetVFolderAction,
     ListVFolderAction,
     MoveToTrashVFolderAction,
+    PurgeVFolderAction,
     RestoreVFolderFromTrashAction,
     UpdateVFolderAttributeAction,
 )
@@ -1937,14 +1938,14 @@ async def purge(request: web.Request, params: PurgeRequestModel) -> web.Response
     ):
         raise InsufficientPrivilege("You are not allowed to purge vfolders")
 
-    async with root_ctx.db.begin_session() as db_session:
-        row = await db_session.scalar(sa.select(VFolderRow).where(VFolderRow.id == folder_id))
-        row = cast(VFolderRow | None, row)
-        if row is None:
-            raise VFolderNotFound(extra_data=folder_id)
-        await check_vfolder_status({"status": row.status}, VFolderStatusSet.PURGABLE)
-        delete_stmt = sa.delete(VFolderRow).where(VFolderRow.id == folder_id)
-        await db_session.execute(delete_stmt)
+    try:
+        await root_ctx.processors.vfolder.purge_vfolder.wait_for_complete(
+            PurgeVFolderAction(
+                purger=Purger(row_class=VFolderRow, pk_value=folder_id),
+            )
+        )
+    except VFolderInvalidParameter as e:
+        raise InvalidAPIParameters(str(e))
 
     return web.Response(status=HTTPStatus.NO_CONTENT)
 

src/ai/backend/manager/repositories/vfolder/repository.py

@@ -58,6 +58,7 @@
     vfolders,
 )
 from ai.backend.manager.repositories.base.creator import Creator
+from ai.backend.manager.repositories.base.purger import Purger, execute_purger
 from ai.backend.manager.repositories.base.updater import Updater, execute_updater
 from ai.backend.manager.repositories.permission_controller.creators import (
     AssociationScopesEntitiesCreatorSpec,
@@ -412,6 +413,19 @@ async def delete_vfolders_forever(self, vfolder_ids: list[uuid.UUID]) -> list[VF
 
             return [self._vfolder_row_to_data(row) for row in vfolder_rows]
 
+
+    @vfolder_repository_resilience.apply()
+    async def purge_vfolder(self, purger: Purger[VFolderRow]) -> VFolderData:
+        """
+        Permanently delete a VFolder from DB.
+        This should only be called for VFolders with DELETE_COMPLETE status.
+        """
+        async with self._db.begin_session() as session:
+            result = await execute_purger(session, purger)
+            if result is None:
+                raise VFolderNotFound(extra_data=str(purger.pk_value))
+            return self._vfolder_row_to_data(result.row)
+
     @vfolder_repository_resilience.apply()
     async def get_vfolder_permissions(self, vfolder_id: uuid.UUID) -> list[VFolderPermissionData]:
         """

src/ai/backend/manager/services/vfolder/actions/base.py

@@ -25,6 +25,7 @@
     VFolderPermission,
     VFolderRow,
 )
+from ai.backend.manager.repositories.base.purger import Purger
 from ai.backend.manager.repositories.base.updater import Updater
 from ai.backend.manager.services.vfolder.types import (
     VFolderBaseInfo,
@@ -382,6 +383,42 @@ def target_entity_id(self) -> str:
         return str(self.vfolder_uuid)
 
 
+@dataclass
+class PurgeVFolderAction(VFolderSingleEntityAction):
+    purger: Purger[VFolderRow]
+
+    @override
+    def entity_id(self) -> Optional[str]:
+        return str(self.purger.pk_value)
+
+    @override
+    @classmethod
+    def operation_type(cls) -> str:
+        return "purge"
+
+    @override
+    @classmethod
+    def permission_operation_type(cls) -> OperationType:
+        return OperationType.HARD_DELETE
+
+    @override
+    def target_entity_id(self) -> str:
+        return str(self.purger.pk_value)
+
+
+@dataclass
+class PurgeVFolderActionResult(VFolderSingleEntityActionResult):
+    vfolder_uuid: uuid.UUID
+
+    @override
+    def entity_id(self) -> Optional[str]:
+        return str(self.vfolder_uuid)
+
+    @override
+    def target_entity_id(self) -> str:
+        return str(self.vfolder_uuid)
+
+
 @dataclass
 class ForceDeleteVFolderAction(VFolderSingleEntityAction):
     """

src/ai/backend/manager/services/vfolder/processors/vfolder.py

@@ -21,6 +21,8 @@
     ListVFolderActionResult,
     MoveToTrashVFolderAction,
     MoveToTrashVFolderActionResult,
+    PurgeVFolderAction,
+    PurgeVFolderActionResult,
     RestoreVFolderFromTrashAction,
     RestoreVFolderFromTrashActionResult,
     UpdateVFolderAttributeAction,
@@ -45,6 +47,7 @@ class VFolderProcessors(AbstractProcessorPackage):
     delete_forever_vfolder: SingleEntityActionProcessor[
         DeleteForeverVFolderAction, DeleteForeverVFolderActionResult
     ]
+    purge_vfolder: SingleEntityActionProcessor[PurgeVFolderAction, PurgeVFolderActionResult]
     force_delete_vfolder: SingleEntityActionProcessor[
         ForceDeleteVFolderAction, ForceDeleteVFolderActionResult
     ]
@@ -67,6 +70,7 @@ def __init__(self, service: VFolderService, action_monitors: list[ActionMonitor]
         self.delete_forever_vfolder = SingleEntityActionProcessor(
             service.delete_forever, action_monitors
         )
+        self.purge_vfolder = SingleEntityActionProcessor(service.purge, action_monitors)
         self.force_delete_vfolder = SingleEntityActionProcessor(
             service.force_delete, action_monitors
         )
@@ -83,6 +87,7 @@ def supported_actions(self) -> list[ActionSpec]:
             MoveToTrashVFolderAction.spec(),
             RestoreVFolderFromTrashAction.spec(),
             DeleteForeverVFolderAction.spec(),
+            PurgeVFolderAction.spec(),
             ForceDeleteVFolderAction.spec(),
             CloneVFolderAction.spec(),
             GetTaskLogsAction.spec(),

src/ai/backend/manager/services/vfolder/services/vfolder.py

@@ -71,6 +71,8 @@
     ListVFolderActionResult,
     MoveToTrashVFolderAction,
     MoveToTrashVFolderActionResult,
+    PurgeVFolderAction,
+    PurgeVFolderActionResult,
     RestoreVFolderFromTrashAction,
     RestoreVFolderFromTrashActionResult,
     UpdateVFolderAttributeAction,
@@ -532,6 +534,21 @@ async def delete_forever(
         await self._remove_vfolder_from_storage(vfolder_data)
         return DeleteForeverVFolderActionResult(vfolder_uuid=action.vfolder_uuid)
 
+
+    async def purge(self, action: PurgeVFolderAction) -> PurgeVFolderActionResult:
+        """Purge a DELETE_COMPLETE vfolder from DB (admin only)."""
+        vfolder_uuid = cast(uuid.UUID, action.purger.pk_value)
+        vfolder_data = await self._vfolder_repository.get_by_id(vfolder_uuid)
+        if vfolder_data is None:
+            raise VFolderNotFound(extra_data=str(vfolder_uuid))
+        if vfolder_data.status not in VFolderStatusSet.PURGABLE:
+            raise VFolderInvalidParameter(
+                f"Cannot purge vfolder with status {vfolder_data.status}"
+            )
+
+        await self._vfolder_repository.purge_vfolder(action.purger)
+        return PurgeVFolderActionResult(vfolder_uuid=vfolder_uuid)
+
     async def force_delete(
         self, action: ForceDeleteVFolderAction
     ) -> ForceDeleteVFolderActionResult: