Added util for loading file based on the path

backend/workflow_manager/endpoint/source.py

@@ -3,6 +3,7 @@
 import os
 import shutil
 from hashlib import md5, sha256
+from io import BytesIO
 from pathlib import Path
 from typing import Any, Optional
 
@@ -384,6 +385,18 @@ def handle_final_result(
         if connection_type == WorkflowEndpoint.ConnectionType.API:
             results.append({"file": file_name, "result": result})
 
+    def load_file(self, input_file_path: str) -> tuple[str, BytesIO]:
+        connector: ConnectorInstance = self.endpoint.connector_instance
+        connector_settings: dict[str, Any] = connector.connector_metadata
+        source_fs: fsspec.AbstractFileSystem = self.get_fsspec(
+            settings=connector_settings, connector_id=connector.connector_id
+        )
+        with source_fs.open(input_file_path, "rb") as remote_file:
+            file_content = remote_file.read()
+            file_stream = BytesIO(file_content)
+
+        return remote_file.key, file_stream
+
     @classmethod
     def add_input_file_to_api_storage(
         cls, workflow_id: str, execution_id: str, file_objs: list[UploadedFile]