feat: support collection export (#1448)

Author: earayu

aperag/api/components/schemas/export.yaml

@@ -0,0 +1,52 @@
+exportTaskResponse:
+  type: object
+  required:
+    - export_task_id
+    - status
+  properties:
+    export_task_id:
+      type: string
+      description: Unique ID of the export task
+    status:
+      type: string
+      enum:
+        - PENDING
+        - PROCESSING
+        - COMPLETED
+        - FAILED
+        - EXPIRED
+      description: Current status of the export task
+    progress:
+      type: integer
+      minimum: 0
+      maximum: 100
+      description: Progress percentage (0-100)
+    message:
+      type: string
+      nullable: true
+      description: Human-readable status message
+    error_message:
+      type: string
+      nullable: true
+      description: Error detail when status is FAILED
+    download_url:
+      type: string
+      nullable: true
+      description: URL to download the ZIP file (only set when status is COMPLETED)
+    file_size:
+      type: integer
+      nullable: true
+      description: Size of the ZIP file in bytes
+    gmt_created:
+      type: string
+      format: date-time
+      nullable: true
+    gmt_completed:
+      type: string
+      format: date-time
+      nullable: true
+    gmt_expires:
+      type: string
+      format: date-time
+      nullable: true
+      description: Time when the export file will be automatically deleted (7 days after creation)

aperag/api/openapi.yaml

@@ -238,3 +238,11 @@ paths:
 
   /agent/message:
     $ref: './paths/agent.yaml'
+
+  # export
+  /collections/{collection_id}/export:
+    $ref: './paths/export.yaml#/createExportTask'
+  /export-tasks/{task_id}:
+    $ref: './paths/export.yaml#/getExportTask'
+  /export-tasks/{task_id}/download:
+    $ref: './paths/export.yaml#/downloadExport'

aperag/api/paths/export.yaml

@@ -0,0 +1,110 @@
+createExportTask:
+  post:
+    summary: Create a knowledge base export task
+    description: |
+      Asynchronously package all object-store files under the collection into a ZIP file.
+      Only the collection owner can initiate an export.
+    operationId: createExportTask
+    security:
+      - BearerAuth: []
+    parameters:
+      - name: collection_id
+        in: path
+        required: true
+        schema:
+          type: string
+        description: Collection ID
+    responses:
+      '202':
+        description: Export task created and queued
+        content:
+          application/json:
+            schema:
+              $ref: '../components/schemas/export.yaml#/exportTaskResponse'
+      '403':
+        description: Permission denied (not collection owner)
+        content:
+          application/json:
+            schema:
+              $ref: '../components/schemas/common.yaml#/failResponse'
+      '404':
+        description: Collection not found
+        content:
+          application/json:
+            schema:
+              $ref: '../components/schemas/common.yaml#/failResponse'
+      '429':
+        description: Too many concurrent export tasks
+        content:
+          application/json:
+            schema:
+              $ref: '../components/schemas/common.yaml#/failResponse'
+
+getExportTask:
+  get:
+    summary: Get export task status
+    description: Query the status and progress of an export task.
+    operationId: getExportTask
+    security:
+      - BearerAuth: []
+    parameters:
+      - name: task_id
+        in: path
+        required: true
+        schema:
+          type: string
+        description: Export task ID
+    responses:
+      '200':
+        description: Export task details
+        content:
+          application/json:
+            schema:
+              $ref: '../components/schemas/export.yaml#/exportTaskResponse'
+      '404':
+        description: Export task not found
+        content:
+          application/json:
+            schema:
+              $ref: '../components/schemas/common.yaml#/failResponse'
+
+downloadExport:
+  get:
+    summary: Download export ZIP
+    description: Stream the completed export ZIP file to the client.
+    operationId: downloadExport
+    security:
+      - BearerAuth: []
+    parameters:
+      - name: task_id
+        in: path
+        required: true
+        schema:
+          type: string
+        description: Export task ID
+    responses:
+      '200':
+        description: ZIP file stream
+        content:
+          application/zip:
+            schema:
+              type: string
+              format: binary
+      '400':
+        description: Export task is not ready for download
+        content:
+          application/json:
+            schema:
+              $ref: '../components/schemas/common.yaml#/failResponse'
+      '404':
+        description: Export task not found
+        content:
+          application/json:
+            schema:
+              $ref: '../components/schemas/common.yaml#/failResponse'
+      '410':
+        description: Export file has expired
+        content:
+          application/json:
+            schema:
+              $ref: '../components/schemas/common.yaml#/failResponse'

aperag/app.py

@@ -46,6 +46,7 @@
 from aperag.views.collections import router as collections_router
 from aperag.views.config import router as config_router
 from aperag.views.evaluation import router as evaluation_router
+from aperag.views.export import router as export_router
 from aperag.views.flow import router as flow_router
 from aperag.views.graph import router as graph_router
 from aperag.views.llm import router as llm_router
@@ -98,6 +99,7 @@ async def health_check():
 app.include_router(auth_router, prefix="/api/v1")
 app.include_router(main_router, prefix="/api/v1")
 app.include_router(collections_router, prefix="/api/v1")  # Add collections router
+app.include_router(export_router, prefix="/api/v1")  # Add export router
 app.include_router(api_key_router, prefix="/api/v1")
 app.include_router(audit_router, prefix="/api/v1")  # Add audit router
 app.include_router(flow_router, prefix="/api/v1")

aperag/db/models.py

@@ -1132,6 +1132,39 @@ class Setting(Base):
     gmt_deleted = Column(DateTime(timezone=True), nullable=True)
 
 
+class ExportTaskStatus(str, Enum):
+    PENDING = "PENDING"
+    PROCESSING = "PROCESSING"
+    COMPLETED = "COMPLETED"
+    FAILED = "FAILED"
+    EXPIRED = "EXPIRED"
+
+
+class ExportTask(Base):
+    __tablename__ = "export_task"
+    __table_args__ = (
+        Index("idx_export_task_user_status", "user", "status"),
+        Index("idx_export_task_expires", "status", "gmt_expires"),
+    )
+
+    id = Column(String(24), primary_key=True, default=lambda: "export" + random_id()[:16])
+    user = Column(String(256), nullable=False, index=True)
+    collection_id = Column(String(24), nullable=False, index=True)
+
+    status = Column(EnumColumn(ExportTaskStatus), nullable=False, default=ExportTaskStatus.PENDING)
+    progress = Column(Integer, default=0)
+    message = Column(Text, nullable=True)
+    error_message = Column(Text, nullable=True)
+
+    object_store_path = Column(Text, nullable=True)
+    file_size = Column(BigInteger, nullable=True)
+
+    gmt_created = Column(DateTime(timezone=True), default=utc_now, nullable=False)
+    gmt_updated = Column(DateTime(timezone=True), default=utc_now, nullable=False)
+    gmt_completed = Column(DateTime(timezone=True), nullable=True)
+    gmt_expires = Column(DateTime(timezone=True), nullable=True)
+
+
 class PromptTemplate(Base):
     __tablename__ = "prompt_template"
 

aperag/migration/versions/20260304120000-a1b2c3d4e5f6.py

@@ -0,0 +1,51 @@
+"""add export_task table
+
+Revision ID: a1b2c3d4e5f6
+Revises: 8abbaf1aa10d
+Create Date: 2026-03-04 12:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = 'a1b2c3d4e5f6'
+down_revision: Union[str, None] = '8abbaf1aa10d'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    op.create_table(
+        'export_task',
+        sa.Column('id', sa.String(length=24), nullable=False),
+        sa.Column('user', sa.String(length=256), nullable=False),
+        sa.Column('collection_id', sa.String(length=24), nullable=False),
+        sa.Column('status', sa.String(length=50), nullable=False),
+        sa.Column('progress', sa.Integer(), nullable=True, server_default='0'),
+        sa.Column('message', sa.Text(), nullable=True),
+        sa.Column('error_message', sa.Text(), nullable=True),
+        sa.Column('object_store_path', sa.Text(), nullable=True),
+        sa.Column('file_size', sa.BigInteger(), nullable=True),
+        sa.Column('gmt_created', sa.DateTime(timezone=True), nullable=False, server_default=sa.text('NOW()')),
+        sa.Column('gmt_updated', sa.DateTime(timezone=True), nullable=False, server_default=sa.text('NOW()')),
+        sa.Column('gmt_completed', sa.DateTime(timezone=True), nullable=True),
+        sa.Column('gmt_expires', sa.DateTime(timezone=True), nullable=True),
+        sa.PrimaryKeyConstraint('id', name=op.f('pk_export_task')),
+    )
+
+    op.create_index(op.f('ix_export_task_user'), 'export_task', ['user'], unique=False)
+    op.create_index(op.f('ix_export_task_collection_id'), 'export_task', ['collection_id'], unique=False)
+    op.create_index('idx_export_task_user_status', 'export_task', ['user', 'status'], unique=False)
+    op.create_index('idx_export_task_expires', 'export_task', ['status', 'gmt_expires'], unique=False)
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    op.drop_index('idx_export_task_expires', table_name='export_task')
+    op.drop_index('idx_export_task_user_status', table_name='export_task')
+    op.drop_index(op.f('ix_export_task_collection_id'), table_name='export_task')
+    op.drop_index(op.f('ix_export_task_user'), table_name='export_task')
+    op.drop_table('export_task')

aperag/objectstore/base.py

@@ -111,6 +111,19 @@ def delete_objects_by_prefix(self, path_prefix: str):
         """
         ...
 
+    @abstractmethod
+    def list_objects_by_prefix(self, path_prefix: str) -> list[str]:
+        """
+        Lists all object paths that start with the given prefix.
+
+        Args:
+            path_prefix: The prefix to match.
+
+        Returns:
+            A list of object paths (relative to the store root) that match the prefix.
+        """
+        ...
+
 
 class AsyncObjectStore(ABC):
     """Abstract base class for asynchronous object storage operations."""
@@ -208,6 +221,19 @@ async def delete_objects_by_prefix(self, path_prefix: str):
         """
         ...
 
+    @abstractmethod
+    async def list_objects_by_prefix(self, path_prefix: str) -> list[str]:
+        """
+        Asynchronously lists all object paths that start with the given prefix.
+
+        Args:
+            path_prefix: The prefix to match.
+
+        Returns:
+            A list of object paths (relative to the store root) that match the prefix.
+        """
+        ...
+
 
 def get_object_store() -> ObjectStore:
     """

aperag/objectstore/local.py

@@ -343,6 +343,23 @@ def delete_objects_by_prefix(self, path_prefix: str):
             logger.error(f"Error during deletion of objects with prefix '{path_prefix}': {e}")
             raise IOError(f"Error during deletion of objects with prefix '{path_prefix}'") from e
 
+    def list_objects_by_prefix(self, path_prefix: str) -> list[str]:
+        normalized_prefix = path_prefix.lstrip("/").replace("\\", "/")
+        result = []
+        try:
+            for item_path in self._base_storage_path.rglob("*"):
+                if item_path.is_file():
+                    try:
+                        relative = str(item_path.relative_to(self._base_storage_path)).replace("\\", "/")
+                        if relative.startswith(normalized_prefix):
+                            result.append(relative)
+                    except ValueError:
+                        logger.debug(f"Item {item_path} not relative to {self._base_storage_path}, skipping.")
+        except Exception as e:
+            logger.error(f"Error listing objects with prefix '{path_prefix}': {e}")
+            raise IOError(f"Error listing objects with prefix '{path_prefix}'") from e
+        return result
+
 
 class AsyncLocal(AsyncObjectStore):
     """Asynchronous wrapper for the Local object store."""
@@ -412,3 +429,6 @@ async def delete(self, path: str):
 
     async def delete_objects_by_prefix(self, path_prefix: str):
         return await sync_to_async(self._sync_store.delete_objects_by_prefix)(path_prefix=path_prefix)
+
+    async def list_objects_by_prefix(self, path_prefix: str) -> list[str]:
+        return await sync_to_async(self._sync_store.list_objects_by_prefix)(path_prefix=path_prefix)