feat/expand error handling to fsspec connectors (#319)

* provide better error handling in s3

* reuse wrap error logic

* provide better error handling in azure

* provide better error handling in dropbox

* provide better error handling in gcs

* provide better error handling in box

* bump changelog

Author: rbiseck3

CHANGELOG.md

@@ -1,3 +1,5 @@
+## 0.3.13-dev0
+
 ## 0.3.12
 
 ### Enhancements

test/integration/connectors/test_s3.py

@@ -16,9 +16,7 @@
     source_connector_validation,
 )
 from test.integration.utils import requires_env
-from unstructured_ingest.error import (
-    SourceConnectionError,
-)
+from unstructured_ingest.v2.errors import UserAuthError, UserError
 from unstructured_ingest.v2.interfaces import FileData, SourceIdentifiers
 from unstructured_ingest.v2.processes.connectors.fsspec.s3 import (
     CONNECTOR_TYPE,
@@ -94,12 +92,19 @@ async def test_s3_source_special_char(anon_connection_config: S3ConnectionConfig
         )
 
 
-@pytest.mark.asyncio
 @pytest.mark.tags(CONNECTOR_TYPE, SOURCE_TAG)
-async def test_s3_source_no_access(anon_connection_config: S3ConnectionConfig):
+def test_s3_source_no_access(anon_connection_config: S3ConnectionConfig):
     indexer_config = S3IndexerConfig(remote_url="s3://utic-ingest-test-fixtures/destination/")
     indexer = S3Indexer(connection_config=anon_connection_config, index_config=indexer_config)
-    with pytest.raises(SourceConnectionError):
+    with pytest.raises(UserAuthError):
+        indexer.precheck()
+
+
+@pytest.mark.tags(CONNECTOR_TYPE, SOURCE_TAG)
+def test_s3_source_no_bucket(anon_connection_config: S3ConnectionConfig):
+    indexer_config = S3IndexerConfig(remote_url="s3://fake-bucket")
+    indexer = S3Indexer(connection_config=anon_connection_config, index_config=indexer_config)
+    with pytest.raises(UserError):
         indexer.precheck()
 
 

unstructured_ingest/__version__.py

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

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

@@ -8,7 +8,9 @@
 from pydantic import Field, Secret
 
 from unstructured_ingest.utils.dep_check import requires_dependencies
+from unstructured_ingest.v2.errors import ProviderError, UserAuthError, UserError
 from unstructured_ingest.v2.interfaces import FileDataSourceMetadata
+from unstructured_ingest.v2.logger import logger
 from unstructured_ingest.v2.processes.connector_registry import (
     DestinationRegistryEntry,
     SourceRegistryEntry,
@@ -98,6 +100,24 @@ def get_client(self, protocol: str) -> Generator["AzureBlobFileSystem", None, No
         with super().get_client(protocol=protocol) as client:
             yield client
 
+    def wrap_error(self, e: Exception) -> Exception:
+        from azure.core.exceptions import ClientAuthenticationError, HttpResponseError
+
+        if not isinstance(e, HttpResponseError):
+            logger.error(f"unhandled exception from azure ({type(e)}): {e}", exc_info=True)
+            return e
+        if isinstance(e, ClientAuthenticationError):
+            return UserAuthError(e.reason)
+        status_code = e.status_code
+        message = e.reason
+        if status_code is not None:
+            if 400 <= status_code < 500:
+                return UserError(message)
+            if status_code >= 500:
+                return ProviderError(message)
+        logger.error(f"unhandled exception from azure ({type(e)}): {e}", exc_info=True)
+        return e
+
 
 @dataclass
 class AzureIndexer(FsspecIndexer):

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

@@ -10,7 +10,9 @@
 from pydantic.functional_validators import BeforeValidator
 
 from unstructured_ingest.utils.dep_check import requires_dependencies
+from unstructured_ingest.v2.errors import ProviderError, UserAuthError, UserError
 from unstructured_ingest.v2.interfaces import FileDataSourceMetadata
+from unstructured_ingest.v2.logger import logger
 from unstructured_ingest.v2.processes.connector_registry import (
     DestinationRegistryEntry,
     SourceRegistryEntry,
@@ -24,7 +26,6 @@
     FsspecIndexerConfig,
     FsspecUploader,
     FsspecUploaderConfig,
-    SourceConnectionError,
 )
 from unstructured_ingest.v2.processes.connectors.utils import conform_string_to_dict
 
@@ -57,13 +58,10 @@ def get_access_config(self) -> dict[str, Any]:
 
         # Create and authenticate the JWTAuth object
         oauth = JWTAuth.from_settings_dictionary(settings_dict)
-        try:
-            oauth.authenticate_instance()
-        except Exception as e:
-            raise SourceConnectionError(f"Failed to authenticate with Box: {e}")
+        oauth.authenticate_instance()
 
-        if not oauth.access_token:
-            raise SourceConnectionError("Authentication failed: No access token generated.")
+        # if not oauth.access_token:
+        #     raise SourceConnectionError("Authentication failed: No access token generated.")
 
         # Prepare the access configuration with the authenticated oauth
         access_kwargs_with_oauth: dict[str, Any] = {
@@ -75,6 +73,24 @@ def get_access_config(self) -> dict[str, Any]:
 
         return access_kwargs_with_oauth
 
+    def wrap_error(self, e: Exception) -> Exception:
+        from boxsdk.exception import BoxAPIException, BoxOAuthException
+
+        if isinstance(e, BoxOAuthException):
+            return UserAuthError(e.message)
+        if not isinstance(e, BoxAPIException):
+            logger.error(f"unhandled exception from box ({type(e)}): {e}", exc_info=True)
+            return e
+        message = e.message or e
+        if error_code_status := e.status:
+            if 400 <= error_code_status < 500:
+                return UserError(message)
+            if error_code_status >= 500:
+                return ProviderError(message)
+
+        logger.error(f"unhandled exception from box ({type(e)}): {e}", exc_info=True)
+        return e
+
     @requires_dependencies(["boxfs"], extras="box")
     @contextmanager
     def get_client(self, protocol: str) -> Generator["BoxFileSystem", None, None]:

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

@@ -8,7 +8,16 @@
 from pydantic import Field, Secret
 
 from unstructured_ingest.utils.dep_check import requires_dependencies
+from unstructured_ingest.v2.errors import (
+    ProviderError,
+    UserAuthError,
+    UserError,
+)
+from unstructured_ingest.v2.errors import (
+    RateLimitError as CustomRateLimitError,
+)
 from unstructured_ingest.v2.interfaces import FileDataSourceMetadata
+from unstructured_ingest.v2.logger import logger
 from unstructured_ingest.v2.processes.connector_registry import (
     DestinationRegistryEntry,
     SourceRegistryEntry,
@@ -53,6 +62,30 @@ def get_client(self, protocol: str) -> Generator["DropboxDriveFileSystem", None,
         with super().get_client(protocol=protocol) as client:
             yield client
 
+    def wrap_error(self, e: Exception) -> Exception:
+        from dropbox.exceptions import AuthError, HttpError, RateLimitError
+
+        if not isinstance(e, HttpError):
+            logger.error(f"unhandled exception from dropbox ({type(e)}): {e}", exc_info=True)
+            return e
+        if isinstance(e, AuthError):
+            raise UserAuthError(e.error)
+        if isinstance(e, RateLimitError):
+            return CustomRateLimitError(e.error)
+        status_code = e.status_code
+        if 400 <= status_code < 500:
+            if body := getattr(e, "body", None):
+                return UserError(body)
+            else:
+                return UserError(e.body)
+        if status_code >= 500:
+            if body := getattr(e, "body", None):
+                return ProviderError(body)
+            else:
+                return ProviderError(e.body)
+        logger.error(f"unhandled exception from dropbox ({type(e)}): {e}", exc_info=True)
+        return e
+
 
 @dataclass
 class DropboxIndexer(FsspecIndexer):

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

@@ -12,11 +12,6 @@
 
 from pydantic import BaseModel, Field, Secret
 
-from unstructured_ingest.error import (
-    DestinationConnectionError,
-    SourceConnectionError,
-    SourceConnectionNetworkError,
-)
 from unstructured_ingest.v2.interfaces import (
     AccessConfig,
     ConnectionConfig,
@@ -88,6 +83,9 @@ def get_client(self, protocol: str) -> Generator["AbstractFileSystem", None, Non
         )
         yield client
 
+    def wrap_error(self, e: Exception) -> Exception:
+        return e
+
 
 FsspecIndexerConfigT = TypeVar("FsspecIndexerConfigT", bound=FsspecIndexerConfig)
 FsspecConnectionConfigT = TypeVar("FsspecConnectionConfigT", bound=FsspecConnectionConfig)
@@ -99,6 +97,9 @@ class FsspecIndexer(Indexer):
     index_config: FsspecIndexerConfigT
     connector_type: str = Field(default=CONNECTOR_TYPE, init=False)
 
+    def wrap_error(self, e: Exception) -> Exception:
+        return self.connection_config.wrap_error(e=e)
+
     def precheck(self) -> None:
         from fsspec import get_filesystem_class
 
@@ -116,7 +117,7 @@ def precheck(self) -> None:
                 client.head(path=file_to_sample)
         except Exception as e:
             logger.error(f"failed to validate connection: {e}", exc_info=True)
-            raise SourceConnectionError(f"failed to validate connection: {e}")
+            raise self.wrap_error(e=e)
 
     def get_file_data(self) -> list[dict[str, Any]]:
         if not self.index_config.recursive:
@@ -230,6 +231,9 @@ def handle_directory_download(self, lpath: Path) -> None:
             shutil.rmtree(lpath)
             shutil.move(src=temp_location, dst=lpath)
 
+    def wrap_error(self, e: Exception) -> Exception:
+        return self.connection_config.wrap_error(e=e)
+
     def run(self, file_data: FileData, **kwargs: Any) -> DownloadResponse:
         download_path = self.get_download_path(file_data=file_data)
         download_path.parent.mkdir(parents=True, exist_ok=True)
@@ -239,8 +243,7 @@ def run(self, file_data: FileData, **kwargs: Any) -> DownloadResponse:
                 client.get(rpath=rpath, lpath=download_path.as_posix())
             self.handle_directory_download(lpath=download_path)
         except Exception as e:
-            logger.error(f"failed to download file {file_data.identifier}: {e}", exc_info=True)
-            raise SourceConnectionNetworkError(f"failed to download file {file_data.identifier}")
+            raise self.wrap_error(e=e)
         return self.generate_download_response(file_data=file_data, download_path=download_path)
 
     async def async_run(self, file_data: FileData, **kwargs: Any) -> DownloadResponse:
@@ -252,8 +255,7 @@ async def async_run(self, file_data: FileData, **kwargs: Any) -> DownloadRespons
                 await client.get(rpath=rpath, lpath=download_path.as_posix())
             self.handle_directory_download(lpath=download_path)
         except Exception as e:
-            logger.error(f"failed to download file {file_data.identifier}: {e}", exc_info=True)
-            raise SourceConnectionNetworkError(f"failed to download file {file_data.identifier}")
+            raise self.wrap_error(e=e)
         return self.generate_download_response(file_data=file_data, download_path=download_path)
 
 
@@ -291,6 +293,9 @@ def __post_init__(self):
                 f"missing 1 required positional argument: 'upload_config'"
             )
 
+    def wrap_error(self, e: Exception) -> Exception:
+        return self.connection_config.wrap_error(e=e)
+
     def precheck(self) -> None:
         from fsspec import get_filesystem_class
 
@@ -301,8 +306,7 @@ def precheck(self) -> None:
             upload_path = Path(self.upload_config.path_without_protocol) / "_empty"
             fs.write_bytes(path=upload_path.as_posix(), value=b"")
         except Exception as e:
-            logger.error(f"failed to validate connection: {e}", exc_info=True)
-            raise DestinationConnectionError(f"failed to validate connection: {e}")
+            raise self.wrap_error(e=e)
 
     def get_upload_path(self, file_data: FileData) -> Path:
         upload_path = (

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

@@ -11,7 +11,9 @@
 
 from unstructured_ingest.utils.dep_check import requires_dependencies
 from unstructured_ingest.utils.string_and_date_utils import json_to_dict
+from unstructured_ingest.v2.errors import ProviderError, UserError
 from unstructured_ingest.v2.interfaces import FileDataSourceMetadata
+from unstructured_ingest.v2.logger import logger
 from unstructured_ingest.v2.processes.connector_registry import (
     DestinationRegistryEntry,
     SourceRegistryEntry,
@@ -103,6 +105,25 @@ def get_client(self, protocol: str) -> Generator["GCSFileSystem", None, None]:
         with super().get_client(protocol=protocol) as client:
             yield client
 
+    def wrap_error(self, e: Exception) -> Exception:
+        # https://github.com/fsspec/gcsfs/blob/main/gcsfs/retry.py#L79
+        from gcsfs.retry import HttpError
+
+        if isinstance(e, FileNotFoundError):
+            raise UserError(f"File not found: {e}")
+        if isinstance(e, OSError) and "Forbidden" in str(e):
+            raise UserError(e)
+        if isinstance(e, ValueError) and "Bad Request" in str(e):
+            raise UserError(e)
+        if isinstance(e, HttpError) and (http_error_code := e.code):
+            message = e.message or e
+            if 400 <= http_error_code < 500:
+                raise UserError(message)
+            if http_error_code >= 500:
+                raise ProviderError(message)
+        logger.error(f"unhandled exception from gcs ({type(e)}): {e}", exc_info=True)
+        return e
+
 
 @dataclass
 class GcsIndexer(FsspecIndexer):

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

@@ -7,9 +7,11 @@
 from pydantic import Field, Secret
 
 from unstructured_ingest.utils.dep_check import requires_dependencies
+from unstructured_ingest.v2.errors import ProviderError, UserAuthError, UserError
 from unstructured_ingest.v2.interfaces import (
     FileDataSourceMetadata,
 )
+from unstructured_ingest.v2.logger import logger
 from unstructured_ingest.v2.processes.connector_registry import (
     DestinationRegistryEntry,
     SourceRegistryEntry,
@@ -79,13 +81,35 @@ def get_client(self, protocol: str) -> Generator["S3FileSystem", None, None]:
         with super().get_client(protocol=protocol) as client:
             yield client
 
+    def wrap_error(self, e: Exception) -> Exception:
+        # s3fs maps botocore errors into python ones using mapping here:
+        # https://github.com/fsspec/s3fs/blob/main/s3fs/errors.py
+        if isinstance(e, PermissionError):
+            return UserAuthError(e)
+        if isinstance(e, FileNotFoundError):
+            return UserError(e)
+        if cause := getattr(e, "__cause__", None):
+            error_response = cause.response
+            error_meta = error_response["ResponseMetadata"]
+            http_code = error_meta["HTTPStatusCode"]
+            message = error_response["Error"].get("Message", str(e))
+            if 400 <= http_code < 500:
+                return UserError(message)
+            if http_code >= 500:
+                return ProviderError(message)
+        logger.error(f"unhandled exception from s3 ({type(e)}): {e}", exc_info=True)
+        return e
+
 
 @dataclass
 class S3Indexer(FsspecIndexer):
     connection_config: S3ConnectionConfig
     index_config: S3IndexerConfig
     connector_type: str = CONNECTOR_TYPE
 
+    def wrap_error(self, e: Exception) -> Exception:
+        return self.connection_config.wrap_error(e=e)
+
     def get_path(self, file_data: dict) -> str:
         return file_data["Key"]