added acl support

Author: lucaseduoli

src/api/router.py

@@ -54,6 +54,7 @@ async def upload_ingest_router(
             session_manager=session_manager,
             task_service=task_service,
             user=user,
+            settings_json=settings_json,
         )
 
     logger.debug("Routing to Langflow upload-ingest pipeline via task service")
@@ -78,12 +79,20 @@ async def _traditional_upload_ingest_task(
     session_manager,
     task_service,
     user: User,
+    settings_json: str | None = None,
 ):
     """Task-based traditional upload and ingest for single/multiple files"""
     try:
         if not upload_files:
             return JSONResponse({"error": "Missing files"}, status_code=400)
 
+        settings = None
+        if settings_json:
+            try:
+                settings = json.loads(settings_json)
+            except json.JSONDecodeError as e:
+                return JSONResponse({"error": f"Invalid settings JSON: {e}"}, status_code=400)
+
         user_id = user.user_id
         user_name = user.name
         user_email = user.email
@@ -121,6 +130,7 @@ async def _traditional_upload_ingest_task(
                 owner_email=user_email,
                 original_filenames=file_path_to_original_filename,
                 replace_duplicates=replace_duplicates,
+                settings=settings,
             )
 
             return JSONResponse(

src/models/processors.py

@@ -190,7 +190,7 @@ async def process_document_standard(
         chunk_size: int = None,
         chunk_overlap: int = None,
         is_sample_data: bool = False,
-        acl: "DocumentACL" = None,
+        acl: "DocumentACL | None" = None,
     ):
         """
         Standard processing pipeline for non-Langflow processors:
@@ -404,6 +404,7 @@ def __init__(
         docling_service=None,
         replace_duplicates: bool = False,
         session_manager=None,
+        settings: dict | None = None,
     ):
         super().__init__(
             document_service,
@@ -421,6 +422,7 @@ def __init__(
         self.session_manager = session_manager or (
             document_service.session_manager if document_service else None
         )
+        self.settings = settings
         if self.session_manager is None:
             raise ValueError("session_manager is required for DocumentFileProcessor")
 
@@ -480,6 +482,17 @@ async def process_item(self, upload_task: UploadTask, item: str, file_task: File
             except Exception:
                 file_size = 0
 
+            # Parse ACL from settings if present
+            from connectors.base import DocumentACL
+            
+            acl = None
+            if self.settings and (self.settings.get("allowed_users") is not None or self.settings.get("allowed_groups") is not None):
+                acl = DocumentACL(
+                    owner=self.owner_user_id,
+                    allowed_users=self.settings.get("allowed_users", []),
+                    allowed_groups=self.settings.get("allowed_groups", [])
+                )
+
             # Use consolidated standard processing
             result = await self.process_document_standard(
                 file_path=item,
@@ -492,6 +505,7 @@ async def process_item(self, upload_task: UploadTask, item: str, file_task: File
                 file_size=file_size,
                 connector_type=self.connector_type,
                 is_sample_data=self.is_sample_data,
+                acl=acl,
             )
 
             file_task.status = TaskStatus.COMPLETED

src/services/task_service.py

@@ -129,6 +129,7 @@ async def create_upload_task(
         owner_email: str = None,
         original_filenames: dict | None = None,
         replace_duplicates: bool = False,
+        settings: dict | None = None,
     ) -> str:
         """Create a new upload task for bulk file processing"""
         # Use default DocumentFileProcessor with user context
@@ -144,6 +145,7 @@ async def create_upload_task(
             docling_service=self.docling_service,
             replace_duplicates=replace_duplicates,
             session_manager=self.session_manager,
+            settings=settings,
         )
         return await self.create_custom_task(
             user_id,