feat: Support appending files when saving (#10631)

* feat: Support appending files when saving

* Update save_file.py

* Update News Aggregator.json

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

* [autofix.ci] apply automated fixes (attempt 3/3)

* Update save_file.py

* Update save_file.py

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

* [autofix.ci] apply automated fixes

* Overwrite existing file if append mode is true

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

* [autofix.ci] apply automated fixes (attempt 3/3)

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

* [autofix.ci] apply automated fixes (attempt 3/3)

* Update test_mcp_servers_file.py

* [autofix.ci] apply automated fixes

* Update save_file.py

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

* [autofix.ci] apply automated fixes (attempt 3/3)

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

* [autofix.ci] apply automated fixes (attempt 3/3)

---------

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

Author: erichare

src/backend/base/langflow/api/v2/files.py

@@ -69,7 +69,15 @@ async def fetch_file_object(file_id: uuid.UUID, current_user: CurrentActiveUser,
     return file
 
 
-async def save_file_routine(file, storage_service, current_user: CurrentActiveUser, file_content=None, file_name=None):
+async def save_file_routine(
+    file,
+    storage_service,
+    current_user: CurrentActiveUser,
+    file_content=None,
+    file_name=None,
+    *,
+    append: bool = False,
+):
     """Routine to save the file content to the storage service."""
     file_id = uuid.uuid4()
 
@@ -79,7 +87,7 @@ async def save_file_routine(file, storage_service, current_user: CurrentActiveUs
         file_name = file.filename
 
     # Save the file using the storage service.
-    await storage_service.save_file(flow_id=str(current_user.id), file_name=file_name, data=file_content)
+    await storage_service.save_file(flow_id=str(current_user.id), file_name=file_name, data=file_content, append=append)
 
     return file_id, file_name
 
@@ -92,6 +100,8 @@ async def upload_user_file(
     current_user: CurrentActiveUser,
     storage_service: Annotated[StorageService, Depends(get_storage_service)],
     settings_service: Annotated[SettingsService, Depends(get_settings_service)],
+    *,
+    append: bool = False,
 ) -> UploadFileResponse:
     """Upload a file for the current user and track it in the database."""
     # Get the max allowed file size from settings (in MB)
@@ -124,12 +134,25 @@ async def upload_user_file(
         mcp_file = await get_mcp_file(current_user)
         mcp_file_ext = await get_mcp_file(current_user, extension=True)
 
+        # Initialize existing_file for append mode
+        existing_file = None
+
         if new_filename == mcp_file_ext:
             # Check if an existing record exists; if so, delete it to replace with the new one
             existing_mcp_file = await get_file_by_name(mcp_file, current_user, session)
             if existing_mcp_file:
                 await delete_file(existing_mcp_file.id, current_user, session, storage_service)
             unique_filename = new_filename
+        elif append:
+            # In append mode, check if file exists and reuse the same filename
+            existing_file = await get_file_by_name(root_filename, current_user, session)
+            if existing_file:
+                # File exists, append to it by reusing the same filename
+                # Extract the filename from the path
+                unique_filename = existing_file.path.split("/")[-1] if "/" in existing_file.path else existing_file.path
+            else:
+                # File doesn't exist yet, create new one with extension
+                unique_filename = f"{root_filename}.{file_extension}" if file_extension else root_filename
         else:
             # For normal files, ensure unique name by appending a count if necessary
             stmt = select(UserFile).where(
@@ -156,7 +179,7 @@ async def upload_user_file(
         # Read file content and save with unique filename
         try:
             file_id, stored_file_name = await save_file_routine(
-                file, storage_service, current_user, file_name=unique_filename
+                file, storage_service, current_user, file_name=unique_filename, append=append
             )
         except Exception as e:
             raise HTTPException(status_code=500, detail=f"Error saving file: {e}") from e
@@ -167,18 +190,26 @@ async def upload_user_file(
             file_name=stored_file_name,
         )
 
-        # Create a new file record
-        new_file = UserFile(
-            id=file_id,
-            user_id=current_user.id,
-            name=root_filename,
-            path=f"{current_user.id}/{stored_file_name}",
-            size=file_size,
-        )
-        session.add(new_file)
+        # In append mode, update existing file record if it exists
+        if append and existing_file:
+            existing_file.size = file_size
+            session.add(existing_file)
+            await session.commit()
+            await session.refresh(existing_file)
+            new_file = existing_file
+        else:
+            # Create a new file record
+            new_file = UserFile(
+                id=file_id,
+                user_id=current_user.id,
+                name=root_filename,
+                path=f"{current_user.id}/{stored_file_name}",
+                size=file_size,
+            )
+            session.add(new_file)
 
-        await session.commit()
-        await session.refresh(new_file)
+            await session.commit()
+            await session.refresh(new_file)
     except Exception as e:
         # Optionally, you could also delete the file from disk if the DB insert fails.
         raise HTTPException(status_code=500, detail=f"Database error: {e}") from e

src/backend/base/langflow/initial_setup/starter_projects/News Aggregator.json

@@ -17,13 +17,14 @@ def build_full_path(self, flow_id: str, file_name: str) -> str:
         """Build the full path of a file in the local storage."""
         return str(self.data_dir / flow_id / file_name)
 
-    async def save_file(self, flow_id: str, file_name: str, data: bytes) -> None:
+    async def save_file(self, flow_id: str, file_name: str, data: bytes, *, append: bool = False) -> None:
         """Save a file in the local storage.
 
         Args:
             flow_id: The identifier for the flow.
             file_name: The name of the file to be saved.
             data: The byte content of the file.
+            append: If True, append to existing file instead of overwriting.
 
         Raises:
             FileNotFoundError: If the specified flow does not exist.
@@ -35,9 +36,11 @@ async def save_file(self, flow_id: str, file_name: str, data: bytes) -> None:
         file_path = folder_path / file_name
 
         try:
-            async with async_open(str(file_path), "wb") as f:
+            mode = "ab" if append else "wb"
+            async with async_open(str(file_path), mode) as f:
                 await f.write(data)
-            await logger.ainfo(f"File {file_name} saved successfully in flow {flow_id}.")
+            action = "appended to" if append else "saved"
+            await logger.ainfo(f"File {file_name} {action} successfully in flow {flow_id}.")
         except Exception:
             logger.exception(f"Error saving file {file_name} in flow {flow_id}")
             raise

src/backend/base/langflow/services/storage/local.py

@@ -15,20 +15,31 @@ def __init__(self, session_service, settings_service) -> None:
         self.s3_client = boto3.client("s3")
         self.set_ready()
 
-    async def save_file(self, folder: str, file_name: str, data) -> None:
+    async def save_file(self, folder: str, file_name: str, data, *, append: bool = False) -> None:
         """Save a file to the S3 bucket.
 
         Args:
             folder: The folder in the bucket to save the file.
             file_name: The name of the file to be saved.
             data: The byte content of the file.
+            append: If True, append to existing file instead of overwriting.
 
         Raises:
             Exception: If an error occurs during file saving.
         """
         try:
+            if append:
+                # For S3, we need to retrieve existing content and append
+                try:
+                    existing_data = await self.get_file(folder, file_name)
+                    data = existing_data + data
+                except ClientError:
+                    # File doesn't exist, proceed with new data
+                    pass
+
             self.s3_client.put_object(Bucket=self.bucket, Key=f"{folder}/{file_name}", Body=data)
-            await logger.ainfo(f"File {file_name} saved successfully in folder {folder}.")
+            action = "appended to" if append else "saved"
+            await logger.ainfo(f"File {file_name} {action} successfully in folder {folder}.")
         except NoCredentialsError:
             await logger.aexception("Credentials not available for AWS S3.")
             raise

src/backend/base/langflow/services/storage/s3.py

@@ -29,7 +29,7 @@ def set_ready(self) -> None:
         self.ready = True
 
     @abstractmethod
-    async def save_file(self, flow_id: str, file_name: str, data) -> None:
+    async def save_file(self, flow_id: str, file_name: str, data, *, append: bool = False) -> None:
         raise NotImplementedError
 
     @abstractmethod

src/backend/base/langflow/services/storage/service.py

@@ -19,8 +19,12 @@ def __init__(self):
         # key -> bytes
         self._store: dict[str, bytes] = {}
 
-    async def save_file(self, flow_id: str, file_name: str, data: bytes):
-        self._store[f"{flow_id}/{file_name}"] = data
+    async def save_file(self, flow_id: str, file_name: str, data: bytes, *, append: bool = False):
+        key = f"{flow_id}/{file_name}"
+        if append and key in self._store:
+            self._store[key] += data
+        else:
+            self._store[key] = data
 
     async def get_file_size(self, flow_id: str, file_name: str):
         return len(self._store.get(f"{flow_id}/{file_name}", b""))

src/backend/tests/unit/api/v2/test_mcp_servers_file.py

@@ -238,3 +238,34 @@ def test_adjust_path_expands_home(self, component_class):
         result = component._adjust_file_path_with_format(input_path, "csv")
         assert str(result) == str(expected_path)
         assert "~" not in str(result)  # Ensure ~ was expanded
+
+    def test_append_mode_txt_file(self, component_class):
+        """Test append mode for text files."""
+        mock_file = MagicMock()
+        mock_parent = MagicMock()
+        mock_parent.exists.return_value = True
+        mock_file.parent = mock_parent
+        mock_file.expanduser.return_value = mock_file
+        mock_file.exists.return_value = True  # File exists for append
+
+        with patch("lfx.components.files_and_knowledge.save_file.Path") as mock_path:
+            mock_path.return_value = mock_file
+            mock_file.read_text.return_value = "Existing content"
+
+            component = component_class()
+            component.set_attributes(
+                {
+                    "input_type": "Message",
+                    "message": Message(text="New content"),
+                    "file_format": "txt",
+                    "file_path": "./test_output.txt",
+                    "append_mode": True,
+                }
+            )
+
+            result = component.save_to_file()
+
+            # Should read existing content and append new content
+            mock_file.read_text.assert_called_with(encoding="utf-8")
+            mock_file.write_text.assert_called_once_with("Existing content\nNew content", encoding="utf-8")
+            assert "appended to" in result