bugfix/fix data read (#438)

* fix stagers that were only reading json rather than have support for ndjson

* bump version

Author: rbiseck3

CHANGELOG.md

@@ -1,3 +1,9 @@
+## 0.5.18
+
+### Fixes
+
+* **Fix missing support for NDJSON in stagers**
+
 ## 0.5.17
 
 ### Fixes 

unstructured_ingest/__version__.py

@@ -1 +1 @@
-__version__ = "0.5.17"  # pragma: no cover
+__version__ = "0.5.18"  # pragma: no cover

unstructured_ingest/utils/data_prep.py

@@ -192,6 +192,16 @@ def get_data(path: Union[Path, str]) -> list[dict]:
             logger.warning(f"failed to read {path} as parquet: {e}")
 
 
+def get_json_data(path: Path) -> list[dict]:
+    with path.open() as f:
+        if path.suffix == ".json":
+            return json.load(f)
+        elif path.suffix == ".ndjson":
+            return ndjson.load(f)
+        else:
+            raise ValueError(f"Unsupported file type: {path}")
+
+
 def get_data_df(path: Path) -> pd.DataFrame:
     with path.open() as f:
         if path.suffix == ".json":

unstructured_ingest/v2/processes/connectors/delta_table.py

@@ -1,4 +1,3 @@
-import json
 import os
 import traceback
 from dataclasses import dataclass, field
@@ -11,7 +10,7 @@
 from pydantic import Field, Secret
 
 from unstructured_ingest.error import DestinationConnectionError
-from unstructured_ingest.utils.data_prep import get_data_df
+from unstructured_ingest.utils.data_prep import get_data_df, get_json_data
 from unstructured_ingest.utils.dep_check import requires_dependencies
 from unstructured_ingest.utils.table import convert_to_pandas_dataframe
 from unstructured_ingest.v2.interfaces import (
@@ -86,9 +85,7 @@ def run(
         output_filename: str,
         **kwargs: Any,
     ) -> Path:
-        with open(elements_filepath) as elements_file:
-            elements_contents = json.load(elements_file)
-
+        elements_contents = get_json_data(elements_filepath)
         output_path = Path(output_dir) / Path(f"{output_filename}.parquet")
 
         df = convert_to_pandas_dataframe(elements_dict=elements_contents)

unstructured_ingest/v2/processes/connectors/neo4j.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import asyncio
-import json
 import uuid
 from collections import defaultdict
 from contextlib import asynccontextmanager
@@ -14,7 +13,7 @@
 
 from unstructured_ingest.error import DestinationConnectionError
 from unstructured_ingest.logger import logger
-from unstructured_ingest.utils.data_prep import batch_generator
+from unstructured_ingest.utils.data_prep import batch_generator, get_json_data
 from unstructured_ingest.utils.dep_check import requires_dependencies
 from unstructured_ingest.v2.interfaces import (
     AccessConfig,
@@ -97,8 +96,7 @@ def run(  # type: ignore
         output_filename: str,
         **kwargs: Any,
     ) -> Path:
-        with elements_filepath.open() as file:
-            elements = json.load(file)
+        elements = get_json_data(elements_filepath)
 
         nx_graph = self._create_lexical_graph(
             elements, self._create_document_node(file_data=file_data)
@@ -294,8 +292,7 @@ def is_async(self):
         return True
 
     async def run_async(self, path: Path, file_data: FileData, **kwargs) -> None:  # type: ignore
-        with path.open() as file:
-            staged_data = json.load(file)
+        staged_data = get_json_data(path)
 
         graph_data = _GraphData.model_validate(staged_data)
         async with self.connection_config.get_client() as client:

unstructured_ingest/v2/processes/connectors/onedrive.py

@@ -35,6 +35,10 @@
     DestinationRegistryEntry,
     SourceRegistryEntry,
 )
+from unstructured_ingest.v2.processes.utils.blob_storage import (
+    BlobStoreUploadStager,
+    BlobStoreUploadStagerConfig,
+)
 
 if TYPE_CHECKING:
     from office365.graph_client import GraphClient
@@ -428,4 +432,6 @@ def run(self, path: Path, file_data: FileData, **kwargs: Any) -> None:
     connection_config=OnedriveConnectionConfig,
     uploader=OnedriveUploader,
     uploader_config=OnedriveUploaderConfig,
+    upload_stager_config=BlobStoreUploadStagerConfig,
+    upload_stager=BlobStoreUploadStager,
 )