feat/explicit opensearch classes (#14)

* Leverage explicit config classes for opensearch connectors

* Add changelog entry

* prep for new release

* tidy

* remove unused check_connection metho

* add in missing fsspec destination precheck

* add changelog entry

* Add requires_dependencies to prechecks in fsspec connectors

Author: rbiseck3

CHANGELOG.md

@@ -1,8 +1,10 @@
-## 0.0.3-dev0
+## 0.0.3
 
 ### Enhancements
 
 * **Improve documentation** Update the README's.
+* **Explicit Opensearch classes** For the connector registry entries for opensearch, use only opensearch specific classes rather than any elasticsearch ones. 
+* **Add missing fsspec destination precheck** check connection in precheck for all fsspec-based destination connectors
 
 ## 0.0.2
 

unstructured_ingest/__version__.py

@@ -1 +1 @@
-__version__ = "0.0.3-dev0"  # pragma: no cover
+__version__ = "0.0.3"  # pragma: no cover

unstructured_ingest/v2/interfaces/process.py

@@ -17,7 +17,3 @@ def run(self, **kwargs: Any) -> Any:
 
     async def run_async(self, **kwargs: Any) -> Any:
         return self.run(**kwargs)
-
-    def check_connection(self):
-        # If the process requires external connections, run a quick check
-        pass

unstructured_ingest/v2/processes/connectors/fsspec/azure.py

@@ -75,6 +75,10 @@ class AzureIndexer(FsspecIndexer):
     index_config: AzureIndexerConfig
     connector_type: str = CONNECTOR_TYPE
 
+    @requires_dependencies(["adlfs", "fsspec"], extras="azure")
+    def precheck(self) -> None:
+        super().precheck()
+
     def sterilize_info(self, path) -> dict:
         info = self.fs.info(path=path)
         return sterilize_dict(data=info, default=azure_json_serial)
@@ -120,6 +124,10 @@ class AzureUploader(FsspecUploader):
     def __post_init__(self):
         super().__post_init__()
 
+    @requires_dependencies(["adlfs", "fsspec"], extras="azure")
+    def precheck(self) -> None:
+        super().precheck()
+
     @requires_dependencies(["adlfs", "fsspec"], extras="azure")
     def run(self, contents: list[UploadContent], **kwargs: Any) -> None:
         return super().run(contents=contents, **kwargs)

unstructured_ingest/v2/processes/connectors/fsspec/box.py

@@ -70,6 +70,10 @@ class BoxIndexer(FsspecIndexer):
     def run(self, **kwargs: Any) -> Generator[FileData, None, None]:
         return super().run(**kwargs)
 
+    @requires_dependencies(["boxfs"], extras="box")
+    def precheck(self) -> None:
+        super().precheck()
+
 
 @dataclass
 class BoxDownloaderConfig(FsspecDownloaderConfig):
@@ -107,6 +111,10 @@ class BoxUploader(FsspecUploader):
     def __post_init__(self):
         super().__post_init__()
 
+    @requires_dependencies(["boxfs"], extras="box")
+    def precheck(self) -> None:
+        super().precheck()
+
     @requires_dependencies(["boxfs"], extras="box")
     def run(self, contents: list[UploadContent], **kwargs: Any) -> None:
         return super().run(contents=contents, **kwargs)

unstructured_ingest/v2/processes/connectors/fsspec/dropbox.py

@@ -57,6 +57,10 @@ def __post_init__(self):
         if not self.index_config.path_without_protocol.startswith("/"):
             self.index_config.path_without_protocol = "/" + self.index_config.path_without_protocol
 
+    @requires_dependencies(["dropboxdrivefs", "fsspec"], extras="dropbox")
+    def precheck(self) -> None:
+        super().precheck()
+
     @requires_dependencies(["dropboxdrivefs", "fsspec"], extras="dropbox")
     def run(self, **kwargs: Any) -> Generator[FileData, None, None]:
         return super().run(**kwargs)
@@ -106,6 +110,10 @@ class DropboxUploader(FsspecUploader):
     def __post_init__(self):
         super().__post_init__()
 
+    @requires_dependencies(["dropboxdrivefs", "fsspec"], extras="dropbox")
+    def precheck(self) -> None:
+        super().precheck()
+
     @requires_dependencies(["dropboxdrivefs", "fsspec"], extras="dropbox")
     def run(self, contents: list[UploadContent], **kwargs: Any) -> None:
         return super().run(contents=contents, **kwargs)

unstructured_ingest/v2/processes/connectors/fsspec/fsspec.py

@@ -9,7 +9,11 @@
 from uuid import NAMESPACE_DNS, uuid5
 
 from unstructured_ingest.enhanced_dataclass import enhanced_field
-from unstructured_ingest.error import SourceConnectionError, SourceConnectionNetworkError
+from unstructured_ingest.error import (
+    DestinationConnectionError,
+    SourceConnectionError,
+    SourceConnectionNetworkError,
+)
 from unstructured_ingest.v2.interfaces import (
     AccessConfig,
     ConnectionConfig,
@@ -300,6 +304,19 @@ def __post_init__(self):
                 f"missing 1 required positional argument: 'upload_config'"
             )
 
+    def precheck(self) -> None:
+        from fsspec import get_filesystem_class
+
+        try:
+            fs = get_filesystem_class(self.upload_config.protocol)(
+                **self.connection_config.get_access_config(),
+            )
+            root_dir = self.upload_config.path_without_protocol.split("/")[0]
+            fs.ls(path=root_dir, detail=False)
+        except Exception as e:
+            logger.error(f"failed to validate connection: {e}", exc_info=True)
+            raise DestinationConnectionError(f"failed to validate connection: {e}")
+
     def get_upload_path(self, file_data: FileData) -> Path:
         upload_path = (
             Path(self.upload_config.path_without_protocol)

unstructured_ingest/v2/processes/connectors/fsspec/gcs.py

@@ -80,6 +80,10 @@ class GcsIndexer(FsspecIndexer):
     def run(self, **kwargs: Any) -> Generator[FileData, None, None]:
         return super().run(**kwargs)
 
+    @requires_dependencies(["gcsfs", "fsspec"], extras="gcs")
+    def precheck(self) -> None:
+        super().precheck()
+
 
 @dataclass
 class GcsDownloaderConfig(FsspecDownloaderConfig):
@@ -117,6 +121,10 @@ class GcsUploader(FsspecUploader):
     def __post_init__(self):
         super().__post_init__()
 
+    @requires_dependencies(["gcsfs", "fsspec"], extras="gcs")
+    def precheck(self) -> None:
+        super().precheck()
+
     @requires_dependencies(["gcsfs", "fsspec"], extras="gcs")
     def run(self, contents: list[UploadContent], **kwargs: Any) -> None:
         return super().run(contents=contents, **kwargs)

unstructured_ingest/v2/processes/connectors/fsspec/s3.py

@@ -111,6 +111,10 @@ def get_metadata(self, path: str) -> FileDataSourceMetadata:
     def run(self, **kwargs: Any) -> Generator[FileData, None, None]:
         return super().run(**kwargs)
 
+    @requires_dependencies(["s3fs", "fsspec"], extras="s3")
+    def precheck(self) -> None:
+        super().precheck()
+
 
 @dataclass
 class S3DownloaderConfig(FsspecDownloaderConfig):
@@ -144,6 +148,10 @@ class S3Uploader(FsspecUploader):
     connection_config: S3ConnectionConfig
     upload_config: S3UploaderConfig = field(default=None)
 
+    @requires_dependencies(["s3fs", "fsspec"], extras="s3")
+    def precheck(self) -> None:
+        super().precheck()
+
     @requires_dependencies(["s3fs", "fsspec"], extras="s3")
     def __post_init__(self):
         super().__post_init__()

unstructured_ingest/v2/processes/connectors/fsspec/sftp.py

@@ -91,6 +91,10 @@ def run(self, **kwargs: Any) -> Generator[FileData, None, None]:
             file.identifier = new_identifier
             yield file
 
+    @requires_dependencies(["paramiko", "fsspec"], extras="sftp")
+    def precheck(self) -> None:
+        super().precheck()
+
 
 @dataclass
 class SftpDownloaderConfig(FsspecDownloaderConfig):
@@ -142,6 +146,10 @@ class SftpUploader(FsspecUploader):
     def __post_init__(self):
         super().__post_init__()
 
+    @requires_dependencies(["paramiko", "fsspec"], extras="sftp")
+    def precheck(self) -> None:
+        super().precheck()
+
     @requires_dependencies(["paramiko", "fsspec"], extras="sftp")
     def run(self, contents: list[UploadContent], **kwargs: Any) -> None:
         return super().run(contents=contents, **kwargs)