Bugfix/ingest pipeline check (#3303)

### Description
Using a `isinstance` on the destination registry mapping breaks when
inheritance is used for the associated uploader types. This adds a
connector type field to all uploaders so that the entry can be
deterministically fetched when running check for associated stager in
pipeline.

Author: rbiseck3

CHANGELOG.md

@@ -1,4 +1,4 @@
-## 0.14.9-dev7
+## 0.14.9-dev8
 
 ### Enhancements
 

unstructured/__version__.py

@@ -1 +1 @@
-__version__ = "0.14.9-dev7"  # pragma: no cover
+__version__ = "0.14.9-dev8"  # pragma: no cover

unstructured/ingest/v2/interfaces/uploader.py

@@ -26,6 +26,7 @@ class UploadContent:
 @dataclass
 class Uploader(BaseProcess, BaseConnector, ABC):
     upload_config: UploaderConfigT
+    connector_type: str
 
     def is_async(self) -> bool:
         return False

unstructured/ingest/v2/pipeline/pipeline.py

@@ -87,17 +87,8 @@ def check_destination_connector(self):
         # Make sure that if the set destination connector expects a stager, one is also set
         if not self.uploader_step:
             return
-        matching_registry_entry = [
-            v
-            for v in destination_registry.values()
-            if isinstance(self.uploader_step.process, v.uploader)
-        ]
-        if len(matching_registry_entry) > 1:
-            raise ValueError(
-                f"More than one entry found in destination registry "
-                f"for uploader type: {self.uploader_step.process}"
-            )
-        registry_entry = matching_registry_entry[0]
+        uploader_connector_type = self.uploader_step.process.connector_type
+        registry_entry = destination_registry[uploader_connector_type]
         if registry_entry.upload_stager and self.stager_step is None:
             raise ValueError(
                 f"pipeline with uploader type {self.uploader_step.process.__class__.__name__} "

unstructured/ingest/v2/processes/connectors/astra.py

@@ -92,6 +92,7 @@ class AstraUploaderConfig(UploaderConfig):
 class AstraUploader(Uploader):
     connection_config: AstraConnectionConfig
     upload_config: AstraUploaderConfig
+    connector_type: str = CONNECTOR_TYPE
 
     @requires_dependencies(["astrapy"], extras="astra")
     def get_collection(self) -> "AstraDBCollection":

unstructured/ingest/v2/processes/connectors/chroma.py

@@ -114,6 +114,7 @@ class ChromaUploaderConfig(UploaderConfig):
 
 @dataclass
 class ChromaUploader(Uploader):
+    connector_type: str = CONNECTOR_TYPE
     upload_config: ChromaUploaderConfig
     connection_config: ChromaConnectionConfig
     client: Optional["Client"] = field(init=False)

unstructured/ingest/v2/processes/connectors/elasticsearch.py

@@ -327,6 +327,7 @@ class ElasticsearchUploaderConfig(UploaderConfig):
 
 @dataclass
 class ElasticsearchUploader(Uploader):
+    connector_type: str = CONNECTOR_TYPE
     upload_config: ElasticsearchUploaderConfig
     connection_config: ElasticsearchConnectionConfig
 

unstructured/ingest/v2/processes/connectors/fsspec/azure.py

@@ -114,6 +114,7 @@ class AzureUploaderConfig(FsspecUploaderConfig):
 
 @dataclass
 class AzureUploader(FsspecUploader):
+    connector_type: str = CONNECTOR_TYPE
     connection_config: AzureConnectionConfig
     upload_config: AzureUploaderConfig = field(default=None)
 

unstructured/ingest/v2/processes/connectors/fsspec/box.py

@@ -100,7 +100,8 @@ class BoxUploaderConfig(FsspecUploaderConfig):
 
 
 @dataclass
-class BoxUpload(FsspecUploader):
+class BoxUploader(FsspecUploader):
+    connector_type: str = CONNECTOR_TYPE
     connection_config: BoxConnectionConfig
     upload_config: BoxUploaderConfig = field(default=None)
 
@@ -131,6 +132,8 @@ async def run_async(self, path: Path, file_data: FileData, **kwargs: Any) -> Non
 add_destination_entry(
     destination_type=CONNECTOR_TYPE,
     entry=DestinationRegistryEntry(
-        uploader=BoxUpload, uploader_config=BoxUploaderConfig, connection_config=BoxConnectionConfig
+        uploader=BoxUploader,
+        uploader_config=BoxUploaderConfig,
+        connection_config=BoxConnectionConfig,
     ),
 )

unstructured/ingest/v2/processes/connectors/fsspec/dropbox.py

@@ -99,7 +99,8 @@ class DropboxUploaderConfig(FsspecUploaderConfig):
 
 
 @dataclass
-class DropboxUpload(FsspecUploader):
+class DropboxUploader(FsspecUploader):
+    connector_type: str = CONNECTOR_TYPE
     connection_config: DropboxConnectionConfig
     upload_config: DropboxUploaderConfig = field(default=None)
 
@@ -130,7 +131,7 @@ async def run_async(self, path: Path, file_data: FileData, **kwargs: Any) -> Non
 add_destination_entry(
     destination_type=CONNECTOR_TYPE,
     entry=DestinationRegistryEntry(
-        uploader=DropboxUpload,
+        uploader=DropboxUploader,
         uploader_config=DropboxUploaderConfig,
         connection_config=DropboxConnectionConfig,
     ),