restructure write params for fsspec method

unstructured/ingest/connector/azure_cognitive_search.py

@@ -84,7 +84,7 @@ def conform_dict(self, data: dict) -> None:
         if page_number := data.get("metadata", {}).get("page_number"):
             data["metadata"]["page_number"] = str(page_number)
 
-    def write_elements(self, elements: t.List[Element]) -> None:
+    def write_elements(self, elements: t.List[Element], *args, **kwargs) -> None:
         elements_dict = convert_to_dict(elements)
         self.write_dict(json_list=elements_dict)
 

unstructured/ingest/connector/delta_table.py

@@ -167,7 +167,7 @@ class DeltaTableDestinationConnector(BaseDestinationConnector):
     def initialize(self):
         pass
 
-    def write_elements(self, elements: t.List[Element]) -> None:
+    def write_elements(self, elements: t.List[Element], *args, **kwargs) -> None:
         elements_json = [json.dumps(e.to_dict()) for e in elements]
         self.write_dict(json_list=elements_json)
 

unstructured/ingest/connector/fsspec.py

@@ -235,7 +235,15 @@ def initialize(self):
             **self.connector_config.get_access_kwargs(),
         )
 
-    def write_elements(self, elements: t.List[Element]) -> None:
+    def write_elements(
+        self,
+        elements: t.List[Element],
+        filename: t.Optional[str] = None,
+        indent: int = 4,
+        encoding: str = "utf-8",
+        *args,
+        **kwargs,
+    ) -> None:
         from fsspec import AbstractFileSystem, get_filesystem_class
 
         fs: AbstractFileSystem = get_filesystem_class(self.connector_config.protocol)(
@@ -244,12 +252,14 @@ def write_elements(self, elements: t.List[Element]) -> None:
 
         logger.info(f"Writing content using filesystem: {type(fs).__name__}")
 
-        s3_path = self.connector_config.path
+        s3_folder = self.connector_config.path
+
+        s3_output_path = str(PurePath(s3_folder, filename)) if filename else s3_folder
         element_dict = convert_to_dict(elements)
-        with tempfile.NamedTemporaryFile(mode="w+") as tmp_file:
-            json.dump(element_dict, tmp_file)
-            logger.debug(f"Uploading {tmp_file.name} -> {s3_path}")
-            fs.put_file(lpath=tmp_file.name, rpath=s3_path)
+        with tempfile.NamedTemporaryFile(mode="w+", encoding=encoding) as tmp_file:
+            json.dump(element_dict, tmp_file, indent=indent)
+            logger.debug(f"Uploading {tmp_file.name} -> {s3_output_path}")
+            fs.put_file(lpath=tmp_file.name, rpath=s3_output_path)
 
     def write(self, docs: t.List[BaseIngestDoc]) -> None:
         from fsspec import AbstractFileSystem, get_filesystem_class

unstructured/ingest/interfaces.py

@@ -484,7 +484,7 @@ def write(self, docs: t.List[BaseIngestDoc]) -> None:
         pass
 
     @abstractmethod
-    def write_elements(self, elements: t.List[Element]) -> None:
+    def write_elements(self, elements: t.List[Element], *args, **kwargs) -> None:
         pass