bugfix/correctly set output blob path based on input path (#494)

Author: bryan-unstructured

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 1.0.29-dev1
+
+* **Fixed issue in the blob storage destination connector where files with the same name were overwriting each other**
+
 ## 1.0.29
 
 ### Fixes

test/integration/connectors/databricks/test_volumes_native.py

@@ -1,7 +1,7 @@
 import json
 import os
 import uuid
-from contextlib import contextmanager
+from contextlib import contextmanager, suppress
 from dataclasses import dataclass
 from pathlib import Path
 from unittest import mock
@@ -205,33 +205,39 @@ def databricks_destination_context(
         yield client
     finally:
         # Cleanup
-        try:
-            for file in client.files.list_directory_contents(
-                directory_path=_get_volume_path(env_data.catalog, volume, volume_path)
-            ):
-                client.files.delete(file.path)
-            client.files.delete_directory(_get_volume_path(env_data.catalog, volume, volume_path))
-        except NotFound:
-            # Directory was never created, don't need to delete
-            pass
-
-
-def validate_upload(client: WorkspaceClient, catalog: str, volume: str, volume_path: str):
-    files = list(
-        client.files.list_directory_contents(
-            directory_path=_get_volume_path(catalog, volume, volume_path)
-        )
-    )
+        with suppress(NotFound):
+            client.workspace.delete(
+                path=_get_volume_path(env_data.catalog, volume, volume_path), recursive=True
+            )
+
+
+def list_files_recursively(client: WorkspaceClient, path: str):
+    files = []
+    objects = client.files.list_directory_contents(path)
+    for obj in objects:
+        full_path = obj.path
+        if obj.is_directory:
+            files.extend(list_files_recursively(client, full_path))
+        else:
+            files.append(full_path)
+    return files
+
+
+def validate_upload(
+    client: WorkspaceClient, catalog: str, volume: str, volume_path: str, num_files: int
+):
+    files = list_files_recursively(client, _get_volume_path(catalog, volume, volume_path))
 
-    assert len(files) == 1
+    assert len(files) == num_files
 
-    resp = client.files.download(files[0].path)
-    data = json.loads(resp.contents.read())
+    for i in range(num_files):
+        resp = client.files.download(files[i])
+        data = json.loads(resp.contents.read())
 
-    assert len(data) == 22
-    element_types = {v["type"] for v in data}
-    assert len(element_types) == 1
-    assert "CompositeElement" in element_types
+        assert len(data) == 22
+        element_types = {v["type"] for v in data}
+        assert len(element_types) == 1
+        assert "CompositeElement" in element_types
 
 
 @pytest.mark.asyncio
@@ -267,4 +273,52 @@ async def test_volumes_native_destination(upload_file: Path):
             catalog=env_data.catalog,
             volume="test-platform",
             volume_path=volume_path,
+            num_files=1,
+        )
+
+
+@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_same_filenames_different_folder(upload_file: Path):
+    env_data = get_basic_auth_env_data()
+    volume_path = f"databricks-volumes-test-output-{uuid.uuid4()}"
+    file_data_1 = FileData(
+        source_identifiers=SourceIdentifiers(
+            fullpath=f"folder1/{upload_file.name}", filename=upload_file.name
+        ),
+        connector_type=CONNECTOR_TYPE,
+        identifier="mock file data",
+    )
+    file_data_2 = FileData(
+        source_identifiers=SourceIdentifiers(
+            fullpath=f"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_1)
+        uploader.run(path=upload_file, file_data=file_data_2)
+
+        validate_upload(
+            client=workspace_client,
+            catalog=env_data.catalog,
+            volume="test-platform",
+            volume_path=volume_path,
+            num_files=2,
         )

test/integration/connectors/test_onedrive.py

@@ -113,7 +113,7 @@ async def test_onedrive_source(temp_dir):
 
 @pytest.mark.tags(CONNECTOR_TYPE, DESTINATION_TAG, BLOB_STORAGE_TAG)
 @requires_env("MS_CLIENT_CRED", "MS_CLIENT_ID", "MS_TENANT_ID", "MS_USER_PNAME")
-def xtest_onedrive_destination(upload_file: Path, onedrive_test_folder: str):
+def test_onedrive_destination(upload_file: Path, onedrive_test_folder: str):
     """
     Integration test for the OneDrive destination connector.
 
@@ -137,7 +137,7 @@ def xtest_onedrive_destination(upload_file: Path, onedrive_test_folder: str):
 
     file_data = FileData(
         source_identifiers=SourceIdentifiers(
-            fullpath=destination_fullpath,
+            fullpath=upload_file.name,
             filename=upload_file.name,
         ),
         connector_type=CONNECTOR_TYPE,

test/integration/connectors/test_s3.py

@@ -182,3 +182,50 @@ async def test_s3_destination(upload_file: Path):
         assert len(uploaded_files) == 1
     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_same_filename_different_folders(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_1 = FileData(
+        source_identifiers=SourceIdentifiers(
+            fullpath="folder1/" + upload_file.name, filename=upload_file.name
+        ),
+        connector_type=CONNECTOR_TYPE,
+        identifier="mock file data",
+    )
+    file_data_2 = FileData(
+        source_identifiers=SourceIdentifiers(
+            fullpath="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_1)
+            await uploader.run_async(path=upload_file, file_data=file_data_2)
+        else:
+            uploader.run(path=upload_file, file_data=file_data_1)
+            uploader.run(path=upload_file, file_data=file_data_2)
+        uploaded_files = [
+            Path(file) for file in s3fs.ls(path=destination_path) if Path(file).name != "_empty"
+        ]
+        assert len(uploaded_files) == 2
+    finally:
+        s3fs.rm(path=destination_path, recursive=True)

unstructured_ingest/__version__.py

@@ -1 +1 @@
-__version__ = "1.0.29"  # pragma: no cover
+__version__ = "1.0.29-dev1"  # pragma: no cover

unstructured_ingest/processes/connectors/databricks/volumes.py

@@ -196,9 +196,14 @@ class DatabricksVolumesUploader(Uploader, ABC):
     connection_config: DatabricksVolumesConnectionConfig
 
     def get_output_path(self, file_data: FileData) -> str:
-        return os.path.join(
-            self.upload_config.path, f"{file_data.source_identifiers.filename}.json"
-        )
+        if file_data.source_identifiers.fullpath:
+            return os.path.join(
+                self.upload_config.path, f"{file_data.source_identifiers.fullpath}.json"
+            )
+        else:
+            return os.path.join(
+                self.upload_config.path, f"{file_data.source_identifiers.filename}.json"
+            )
 
     def precheck(self) -> None:
         try:

unstructured_ingest/processes/connectors/fsspec/fsspec.py

@@ -344,8 +344,7 @@ def precheck(self) -> None:
 
     def get_upload_path(self, file_data: FileData) -> Path:
         upload_path = (
-            Path(self.upload_config.path_without_protocol)
-            / file_data.source_identifiers.relative_path
+            Path(self.upload_config.path_without_protocol) / file_data.source_identifiers.fullpath
         )
         updated_upload_path = upload_path.parent / f"{upload_path.name}.json"
         return updated_upload_path
@@ -358,8 +357,8 @@ def run(self, path: Path, file_data: FileData, **kwargs: Any) -> None:
             client.upload(lpath=path_str, rpath=upload_path.as_posix())
 
     async def run_async(self, path: Path, file_data: FileData, **kwargs: Any) -> None:
-        upload_path = self.get_upload_path(file_data=file_data)
         path_str = str(path.resolve())
+        upload_path = self.get_upload_path(file_data=file_data)
         # Odd that fsspec doesn't run exists() as async even when client support async
         logger.debug(f"writing local file {path_str} to {upload_path}")
         with self.connection_config.get_client(protocol=self.upload_config.protocol) as client:

unstructured_ingest/processes/connectors/onedrive.py

@@ -370,14 +370,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 relative path to maintain directory structure, if needed
-        if file_data.source_identifiers and file_data.source_identifiers.rel_path:
-            # Combine the base destination folder with the file's relative path
+        # 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
             destination_path = Path(base_destination_folder) / Path(
-                f"{file_data.source_identifiers.rel_path}.json"
+                f"{file_data.source_identifiers.fullpath}.json"
             )
         else:
-            # If no relative path is provided, upload directly to the base destination folder
+            # If no full 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