fix output path in fsspec connector (#528)

the output path in destination connector should use relative path,
instead of full path. If relative path is not available, use filename
instead.

Author: bryan-unstructured

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 1.0.35
+
+* **Fix output path in blob storage destination connector**
+
 ## 1.0.34
 
 * **Improve Confluence Indexer's precheck** - validate access to each space

test/integration/connectors/databricks/test_volumes_native.py

@@ -224,11 +224,18 @@ def list_files_recursively(client: WorkspaceClient, path: str):
 
 
 def validate_upload(
-    client: WorkspaceClient, catalog: str, volume: str, volume_path: str, num_files: int
+    client: WorkspaceClient,
+    catalog: str,
+    volume: str,
+    volume_path: str,
+    num_files: int,
+    filepath_in_destination: list[str] = None,
 ):
     files = list_files_recursively(client, _get_volume_path(catalog, volume, volume_path))
 
     assert len(files) == num_files
+    if filepath_in_destination:
+        assert files == filepath_in_destination
 
     for i in range(num_files):
         resp = client.files.download(files[i])
@@ -322,3 +329,48 @@ async def test_volumes_native_destination_same_filenames_different_folder(upload
             volume_path=volume_path,
             num_files=2,
         )
+
+
+@pytest.mark.asyncio
+@pytest.mark.tags(CONNECTOR_TYPE, DESTINATION_TAG, BLOB_STORAGE_TAG)
+@requires_env(
+    "DATABRICKS_HOST", "DATABRICKS_CLIENT_ID", "DATABRICKS_CLIENT_SECRET", "DATABRICKS_CATALOG"
+)
+async def test_volumes_native_destination_different_fullpath_relative_path(upload_file: Path):
+    env_data = get_basic_auth_env_data()
+    volume_path = f"databricks-volumes-test-output-{uuid.uuid4()}"
+    file_data = FileData(
+        source_identifiers=SourceIdentifiers(
+            relative_path=f"folder2/{upload_file.name}",
+            fullpath=f"folder1/folder2/{upload_file.name}",
+            filename=upload_file.name,
+        ),
+        connector_type=CONNECTOR_TYPE,
+        identifier="mock file data",
+    )
+    with databricks_destination_context(
+        volume="test-platform", volume_path=volume_path, env_data=env_data
+    ) as workspace_client:
+        connection_config = env_data.get_connection_config()
+        uploader = DatabricksNativeVolumesUploader(
+            connection_config=connection_config,
+            upload_config=DatabricksNativeVolumesUploaderConfig(
+                volume="test-platform",
+                volume_path=volume_path,
+                catalog=env_data.catalog,
+            ),
+        )
+        uploader.precheck()
+        uploader.run(path=upload_file, file_data=file_data)
+
+        filepath_in_destination = [
+            f"/Volumes/utic-dev-tech-fixtures/default/test-platform/{volume_path}/folder1/folder2/DA-1p-with-duplicate-pages.pdf.json.json"
+        ]
+        validate_upload(
+            client=workspace_client,
+            catalog=env_data.catalog,
+            volume="test-platform",
+            volume_path=volume_path,
+            num_files=1,
+            filepath_in_destination=filepath_in_destination,
+        )

test/integration/connectors/test_s3.py

@@ -229,3 +229,43 @@ async def test_s3_destination_same_filename_different_folders(upload_file: Path)
         assert len(uploaded_files) == 2
     finally:
         s3fs.rm(path=destination_path, recursive=True)
+
+
+@pytest.mark.asyncio
+@pytest.mark.tags(CONNECTOR_TYPE, DESTINATION_TAG, BLOB_STORAGE_TAG)
+@requires_env("S3_INGEST_TEST_ACCESS_KEY", "S3_INGEST_TEST_SECRET_KEY")
+async def test_s3_destination_different_relative_path_and_full_path(upload_file: Path):
+    aws_credentials = get_aws_credentials()
+    s3_bucket = "s3://utic-ingest-test-fixtures"
+    destination_path = f"{s3_bucket}/destination/{uuid.uuid4()}"
+    connection_config = S3ConnectionConfig(
+        access_config=S3AccessConfig(
+            key=aws_credentials["aws_access_key_id"],
+            secret=aws_credentials["aws_secret_access_key"],
+        ),
+    )
+    upload_config = S3UploaderConfig(remote_url=destination_path)
+    uploader = S3Uploader(connection_config=connection_config, upload_config=upload_config)
+    s3fs = uploader.fs
+    file_data = FileData(
+        source_identifiers=SourceIdentifiers(
+            relative_path=f"folder2/{upload_file.name}",
+            fullpath=f"folder1/folder2/{upload_file.name}",
+            filename=upload_file.name,
+        ),
+        connector_type=CONNECTOR_TYPE,
+        identifier="mock file data",
+    )
+    try:
+        uploader.precheck()
+        if uploader.is_async():
+            await uploader.run_async(path=upload_file, file_data=file_data)
+        else:
+            uploader.run(path=upload_file, file_data=file_data)
+        uploaded_files = [
+            Path(file) for file in s3fs.ls(path=destination_path) if Path(file).name != "_empty"
+        ]
+        assert len(uploaded_files) == 1
+        assert uploaded_files[0].as_posix() == f"{destination_path.lstrip('s3://')}/folder1"
+    finally:
+        s3fs.rm(path=destination_path, recursive=True)

unstructured_ingest/__version__.py

@@ -1 +1 @@
-__version__ = "1.0.34"  # pragma: no cover
+__version__ = "1.0.35"  # pragma: no cover

unstructured_ingest/processes/connectors/databricks/volumes.py

@@ -196,9 +196,10 @@ class DatabricksVolumesUploader(Uploader, ABC):
     connection_config: DatabricksVolumesConnectionConfig
 
     def get_output_path(self, file_data: FileData) -> str:
-        if file_data.source_identifiers.fullpath:
+        if file_data.source_identifiers.relative_path:
             return os.path.join(
-                self.upload_config.path, f"{file_data.source_identifiers.fullpath}.json"
+                self.upload_config.path,
+                f"{file_data.source_identifiers.relative_path.lstrip('/')}.json",
             )
         else:
             return os.path.join(

unstructured_ingest/processes/connectors/fsspec/fsspec.py

@@ -343,9 +343,10 @@ def precheck(self) -> None:
             raise self.wrap_error(e=e)
 
     def get_upload_path(self, file_data: FileData) -> Path:
-        upload_path = Path(
-            self.upload_config.path_without_protocol
-        ) / file_data.source_identifiers.fullpath.lstrip("/")
+        upload_path = (
+            Path(self.upload_config.path_without_protocol)
+            / file_data.source_identifiers.relative_path.lstrip("/")
+        )
         updated_upload_path = upload_path.parent / f"{upload_path.name}.json"
         return updated_upload_path
 

unstructured_ingest/processes/connectors/onedrive.py

@@ -369,15 +369,14 @@ def run(self, path: Path, file_data: FileData, **kwargs: Any) -> None:
 
         # Use the remote_url from upload_config as the base destination folder
         base_destination_folder = self.upload_config.url
-
-        # Use the file's full path to maintain directory structure, if needed
-        if file_data.source_identifiers and file_data.source_identifiers.fullpath:
-            # Combine the base destination folder with the file's full path
+        # Use the file's relative path to maintain directory structure, if needed
+        if file_data.source_identifiers and file_data.source_identifiers.relative_path:
+            # Combine the base destination folder with the file's relative path
             destination_path = Path(base_destination_folder) / Path(
-                f"{file_data.source_identifiers.fullpath}.json"
+                f"{file_data.source_identifiers.relative_path}.json"
             )
         else:
-            # If no full path is provided, upload directly to the base destination folder
+            # If no relative path is provided, upload directly to the base destination folder
             destination_path = Path(base_destination_folder) / f"{path.name}.json"
 
         destination_folder = destination_path.parent