feat: raise errors from document search ingest when any document failed (#483)

Author: mhordynski

packages/ragbits-document-search/CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## Unreleased
 
+- DocumentSearch.ingest now raises IngestExecutionError when any errors are encountered during ingestion.
+
 ## 0.12.0 (2025-03-25)
 
 ### Changed

packages/ragbits-document-search/src/ragbits/document_search/_main.py

@@ -27,7 +27,7 @@
     IngestStrategy,
     SequentialIngestStrategy,
 )
-from ragbits.document_search.ingestion.strategies.base import IngestExecutionResult
+from ragbits.document_search.ingestion.strategies.base import IngestExecutionError, IngestExecutionResult
 from ragbits.document_search.retrieval.rephrasers.base import QueryRephraser
 from ragbits.document_search.retrieval.rephrasers.noop import NoopQueryRephraser
 from ragbits.document_search.retrieval.rerankers.base import Reranker, RerankerOptions
@@ -210,25 +210,36 @@ async def search(self, query: str, config: SearchConfig | None = None) -> Sequen
             return outputs.search_results
 
     @traceable
-    async def ingest(self, documents: str | Iterable[DocumentMeta | Document | Source]) -> IngestExecutionResult:
+    async def ingest(
+        self, documents: str | Iterable[DocumentMeta | Document | Source], fail_on_error: bool = True
+    ) -> IngestExecutionResult:
         """
         Ingest documents into the search index.
 
         Args:
-            documents: Either:
-                - A iterable of `Document`, `DocumentMetadata`, or `Source` objects
-                - A source-specific URI string (e.g., "gcs://bucket/*") to specify source location(s), for example:
-                    - "file:///path/to/files/*.txt"
-                    - "gcs://bucket/folder/*"
-                    - "huggingface://dataset/split/row"
+            documents: A string representing a source-specific URI (e.g., "gcs://bucket/*") or an iterable of
+                       `Document`, `DocumentMeta`, or `Source` objects. Examples of URI formats include:
+                       - "file:///path/to/files/*.txt"
+                       - "gcs://bucket/folder/*"
+                       - "huggingface://dataset/split/row"
+            fail_on_error: If True, raises IngestExecutionError when any errors are encountered during ingestion.
+                           If False, returns all errors encountered in the IngestExecutionResult.
 
         Returns:
-            The ingest execution result.
+            An IngestExecutionResult containing the results of the ingestion process.
+
+        Raises:
+            IngestExecutionError: If fail_on_error is True and any errors are encountered during ingestion.
         """
         resolved_documents = await SourceResolver.resolve(documents) if isinstance(documents, str) else documents
-        return await self.ingest_strategy(
+        results = await self.ingest_strategy(
             documents=resolved_documents,
             vector_store=self.vector_store,
             parser_router=self.parser_router,
             enricher_router=self.enricher_router,
         )
+
+        if fail_on_error and results.failed:
+            raise IngestExecutionError(results.failed)
+
+        return results

packages/ragbits-document-search/src/ragbits/document_search/ingestion/strategies/base.py

@@ -67,6 +67,15 @@ class IngestExecutionResult:
     failed: list[IngestDocumentResult] = field(default_factory=list)
 
 
+class IngestExecutionError(Exception):
+    """
+    Represents an error that occurred during the documents ingest execution.
+    """
+
+    def __init__(self, results: list[IngestDocumentResult]) -> None:
+        self.results = results
+
+
 class IngestStrategy(WithConstructionConfig, ABC):
     """
     Base class for ingest strategies, responsible for orchiesting the tasks required to index the document.

packages/ragbits-document-search/tests/unit/test_document_search_ingest_errors.py

@@ -0,0 +1,68 @@
+import pytest
+
+from ragbits.core.embeddings.noop import NoopEmbedder
+from ragbits.core.vector_stores.in_memory import InMemoryVectorStore
+from ragbits.document_search import DocumentSearch
+from ragbits.document_search.documents.document import Document, DocumentMeta, DocumentType
+from ragbits.document_search.documents.element import Element
+from ragbits.document_search.ingestion.parsers.base import DocumentParser
+from ragbits.document_search.ingestion.parsers.router import DocumentParserRouter
+from ragbits.document_search.ingestion.strategies.base import IngestExecutionError
+
+
+class FailingParser(DocumentParser):
+    """A parser that always raises an exception."""
+
+    supported_document_types = {DocumentType.TXT}
+
+    @classmethod
+    async def parse(cls, document: Document) -> list[Element]:
+        raise ValueError("This parser always fails")
+
+
+async def test_ingest_fails_on_error():
+    # Create a document search instance with a failing parser
+    document_search = DocumentSearch(
+        vector_store=InMemoryVectorStore(embedder=NoopEmbedder()),
+        parser_router=DocumentParserRouter({DocumentType.TXT: FailingParser()}),
+    )
+
+    # Create a test document
+    document = DocumentMeta.create_text_document_from_literal("Test content")
+
+    # Test that ingest raises IngestExecutionError when fail_on_error=True (default)
+    with pytest.raises(IngestExecutionError) as exc_info:
+        await document_search.ingest([document])
+
+    # Verify the error details
+    assert len(exc_info.value.results) == 1
+    failed_result = exc_info.value.results[0]
+    assert failed_result.document_uri == document.id
+    assert failed_result.num_elements == 0
+    assert failed_result.error is not None
+    assert isinstance(failed_result.error.type, type(ValueError))
+    assert failed_result.error.message == "This parser always fails"
+
+
+async def test_ingest_returns_errors_when_fail_on_error_false():
+    # Create a document search instance with a failing parser
+    document_search = DocumentSearch(
+        vector_store=InMemoryVectorStore(embedder=NoopEmbedder()),
+        parser_router=DocumentParserRouter({DocumentType.TXT: FailingParser()}),
+    )
+
+    # Create a test document
+    document = DocumentMeta.create_text_document_from_literal("Test content")
+
+    # Test that ingest returns errors when fail_on_error=False
+    result = await document_search.ingest([document], fail_on_error=False)
+
+    # Verify the result details
+    assert len(result.successful) == 0
+    assert len(result.failed) == 1
+    failed_result = result.failed[0]
+    assert failed_result.document_uri == document.id
+    assert failed_result.num_elements == 0
+    assert failed_result.error is not None
+    assert isinstance(failed_result.error.type, type(ValueError))
+    assert failed_result.error.message == "This parser always fails"