From ed36cc2230ce5dee3d1b0e5eebe0130924b74541 Mon Sep 17 00:00:00 2001 From: Yiorgis Gozadinos Date: Mon, 25 May 2026 12:31:31 +0300 Subject: [PATCH] typed UnsupportedSourceError replaces pipeline string-marker matching --- haiku_rag_slim/haiku/rag/client/documents.py | 11 +++++--- haiku_rag_slim/haiku/rag/client/exceptions.py | 7 ++++++ haiku_rag_slim/haiku/rag/client/processing.py | 13 ++++++---- .../haiku/rag/ingester/sources/registry.py | 5 ++-- .../haiku/rag/ingester/sources/s3.py | 5 ++-- .../haiku/rag/ingester/workers/pipeline.py | 25 ++++++++----------- tests/ingester/test_pipeline.py | 16 +++++++++--- 7 files changed, 51 insertions(+), 31 deletions(-) create mode 100644 haiku_rag_slim/haiku/rag/client/exceptions.py diff --git a/haiku_rag_slim/haiku/rag/client/documents.py b/haiku_rag_slim/haiku/rag/client/documents.py index 27cd1adf..ec5c2d60 100644 --- a/haiku_rag_slim/haiku/rag/client/documents.py +++ b/haiku_rag_slim/haiku/rag/client/documents.py @@ -5,6 +5,7 @@ from urllib.parse import unquote, urlparse import logfire +from haiku.rag.client.exceptions import UnsupportedSourceError from haiku.rag.client.processing import ( ensure_chunks_embedded, get_extension_from_content_type_or_url, @@ -226,7 +227,7 @@ async def _ingest_fetch_result( result.uri, result.content_type ) if file_extension not in converter.supported_extensions: - raise ValueError( + raise UnsupportedSourceError( f"Unsupported content type/extension: {result.content_type}/{file_extension}" ) @@ -338,7 +339,7 @@ async def create_document_from_source( ) if local_path.is_dir(): if uri is not None: - raise ValueError( + raise UnsupportedSourceError( "uri override is not supported for directory sources; each file " "produces its own document with its own auto-derived URI." ) @@ -359,13 +360,15 @@ async def create_document_from_source( return documents if not local_path.exists(): - raise ValueError(f"File does not exist: {local_path}") + raise UnsupportedSourceError(f"File does not exist: {local_path}") # Match the old _create_document_from_file behaviour: fail fast on # unsupported extension before reading any bytes. converter = get_converter(client._config) if local_path.suffix.lower() not in converter.supported_extensions: - raise ValueError(f"Unsupported file extension: {local_path.suffix}") + raise UnsupportedSourceError( + f"Unsupported file extension: {local_path.suffix}" + ) # Single resource — resolve the right Source adapter for this URI. fetcher = resolve_fetcher(source_str, storage_options=storage_options) diff --git a/haiku_rag_slim/haiku/rag/client/exceptions.py b/haiku_rag_slim/haiku/rag/client/exceptions.py new file mode 100644 index 00000000..018435b6 --- /dev/null +++ b/haiku_rag_slim/haiku/rag/client/exceptions.py @@ -0,0 +1,7 @@ +class UnsupportedSourceError(ValueError): + """A source URI or file cannot be ingested and never will be on a retry — + unsupported extension, unsupported content type, missing file, malformed + S3 URI, etc. ``ValueError`` for backward compatibility with callers that + used to catch a plain ValueError; the ingester pipeline catches the + specific type to classify as ``PermanentError`` without string matching. + """ diff --git a/haiku_rag_slim/haiku/rag/client/processing.py b/haiku_rag_slim/haiku/rag/client/processing.py index 0b8583ca..224dbe6c 100644 --- a/haiku_rag_slim/haiku/rag/client/processing.py +++ b/haiku_rag_slim/haiku/rag/client/processing.py @@ -6,6 +6,7 @@ from urllib.parse import urlparse import httpx +from haiku.rag.client.exceptions import UnsupportedSourceError from haiku.rag.config import AppConfig from haiku.rag.converters import get_converter from haiku.rag.store.models.chunk import Chunk @@ -100,9 +101,9 @@ async def convert( # Path object - convert file directly if isinstance(source, Path): if not source.exists(): - raise ValueError(f"File does not exist: {source}") + raise UnsupportedSourceError(f"File does not exist: {source}") if source.suffix.lower() not in converter.supported_extensions: - raise ValueError(f"Unsupported file extension: {source.suffix}") + raise UnsupportedSourceError(f"Unsupported file extension: {source.suffix}") effective_uri = source_uri or source.absolute().as_uri() doc = await _convert_file(source, effective_uri) _warn_if_descriptions_missing(config, doc, str(source)) @@ -123,7 +124,7 @@ async def convert( ) if file_extension not in converter.supported_extensions: - raise ValueError( + raise UnsupportedSourceError( f"Unsupported content type/extension: {content_type}/{file_extension}" ) @@ -146,9 +147,11 @@ async def convert( # file:// URI file_path = Path(parsed.path) if not file_path.exists(): - raise ValueError(f"File does not exist: {file_path}") + raise UnsupportedSourceError(f"File does not exist: {file_path}") if file_path.suffix.lower() not in converter.supported_extensions: - raise ValueError(f"Unsupported file extension: {file_path.suffix}") + raise UnsupportedSourceError( + f"Unsupported file extension: {file_path.suffix}" + ) effective_uri = source_uri or file_path.absolute().as_uri() doc = await _convert_file(file_path, effective_uri) _warn_if_descriptions_missing(config, doc, str(file_path)) diff --git a/haiku_rag_slim/haiku/rag/ingester/sources/registry.py b/haiku_rag_slim/haiku/rag/ingester/sources/registry.py index e005180b..6ed14b97 100644 --- a/haiku_rag_slim/haiku/rag/ingester/sources/registry.py +++ b/haiku_rag_slim/haiku/rag/ingester/sources/registry.py @@ -2,6 +2,7 @@ from collections.abc import Iterable from pathlib import Path from urllib.parse import urlparse +from haiku.rag.client.exceptions import UnsupportedSourceError from haiku.rag.ingester.sources.base import Source from haiku.rag.ingester.sources.fs import FSSource from haiku.rag.ingester.sources.http import HTTPSource @@ -36,7 +37,7 @@ def resolve_fetcher( if scheme == "s3": bucket = urlparse(uri).netloc if not bucket: - raise ValueError(f"Invalid S3 URI: {uri}") + raise UnsupportedSourceError(f"Invalid S3 URI: {uri}") return S3Source(uri=f"s3://{bucket}/", storage_options=storage_options) - raise ValueError(f"No source adapter for URI scheme {scheme!r}: {uri}") + raise UnsupportedSourceError(f"No source adapter for URI scheme {scheme!r}: {uri}") diff --git a/haiku_rag_slim/haiku/rag/ingester/sources/s3.py b/haiku_rag_slim/haiku/rag/ingester/sources/s3.py index a250945a..fe6d6917 100644 --- a/haiku_rag_slim/haiku/rag/ingester/sources/s3.py +++ b/haiku_rag_slim/haiku/rag/ingester/sources/s3.py @@ -4,6 +4,7 @@ from collections.abc import AsyncIterator from datetime import UTC, datetime from urllib.parse import urlparse +from haiku.rag.client.exceptions import UnsupportedSourceError from haiku.rag.ingester.sources.base import ( FetchResult, RevisionSnapshot, @@ -19,7 +20,7 @@ from haiku.rag.ingester.sources.filter import ( def _parse_s3_uri(uri: str) -> tuple[str, str]: parsed = urlparse(uri) if parsed.scheme != "s3" or not parsed.netloc: - raise ValueError(f"Invalid S3 URI: {uri}") + raise UnsupportedSourceError(f"Invalid S3 URI: {uri}") return parsed.netloc, parsed.path.lstrip("/") @@ -27,7 +28,7 @@ def _parse_s3_object_uri(uri: str) -> tuple[str, str]: """Like _parse_s3_uri but rejects bucket-only URIs (no key).""" bucket, key = _parse_s3_uri(uri) if not key: - raise ValueError(f"Invalid S3 URI: {uri}") + raise UnsupportedSourceError(f"Invalid S3 URI: {uri}") return bucket, key diff --git a/haiku_rag_slim/haiku/rag/ingester/workers/pipeline.py b/haiku_rag_slim/haiku/rag/ingester/workers/pipeline.py index d2c52a0c..05aaa3ae 100644 --- a/haiku_rag_slim/haiku/rag/ingester/workers/pipeline.py +++ b/haiku_rag_slim/haiku/rag/ingester/workers/pipeline.py @@ -6,22 +6,13 @@ import httpx import logfire from pydantic import BaseModel +from haiku.rag.client.exceptions import UnsupportedSourceError from haiku.rag.ingester.exceptions import PermanentError, TransientError from haiku.rag.ingester.queue.models import Job, JobOp if TYPE_CHECKING: from haiku.rag.client import HaikuRAG -# ValueError messages from create_document_from_source that mean "this job -# will never succeed". Anything else from ValueError defaults to transient. -_PERMANENT_VALUE_MARKERS = ( - "Unsupported file extension", - "Unsupported content type", - "Invalid S3 URI", - "File does not exist", - "uri override is not supported", -) - class JobResult(BaseModel): """What the worker needs after a successful job: enough metadata to @@ -39,11 +30,17 @@ def _classify(exc: BaseException) -> Exception: if isinstance(exc, PermanentError | TransientError): return exc + # UnsupportedSourceError is the typed signal from client/* that the + # source will never ingest successfully on a retry (bad URI scheme, + # missing file, unsupported extension, etc.). + if isinstance(exc, UnsupportedSourceError): + return PermanentError(str(exc)) + if isinstance(exc, ValueError): - message = str(exc) - if any(marker in message for marker in _PERMANENT_VALUE_MARKERS): - return PermanentError(message) - return TransientError(message) + # Some downstream libraries (e.g. docling) raise plain ValueError + # for "couldn't parse this file"; default to transient so the queue + # retries up to max_attempts in case the issue is intermittent. + return TransientError(str(exc)) if isinstance(exc, httpx.HTTPStatusError): status = exc.response.status_code diff --git a/tests/ingester/test_pipeline.py b/tests/ingester/test_pipeline.py index a9d4f40c..17180384 100644 --- a/tests/ingester/test_pipeline.py +++ b/tests/ingester/test_pipeline.py @@ -104,8 +104,10 @@ async def test_delete_is_noop_when_document_missing(): @pytest.mark.asyncio async def test_unsupported_extension_classified_permanent(): + from haiku.rag.client.exceptions import UnsupportedSourceError + client = _mock_client() - client.create_document_from_source.side_effect = ValueError( + client.create_document_from_source.side_effect = UnsupportedSourceError( "Unsupported file extension: .xyz" ) @@ -115,8 +117,10 @@ async def test_unsupported_extension_classified_permanent(): @pytest.mark.asyncio async def test_unsupported_content_type_classified_permanent(): + from haiku.rag.client.exceptions import UnsupportedSourceError + client = _mock_client() - client.create_document_from_source.side_effect = ValueError( + client.create_document_from_source.side_effect = UnsupportedSourceError( "Unsupported content type/extension: application/octet-stream/.bin" ) with pytest.raises(PermanentError): @@ -125,8 +129,10 @@ async def test_unsupported_content_type_classified_permanent(): @pytest.mark.asyncio async def test_invalid_s3_uri_classified_permanent(): + from haiku.rag.client.exceptions import UnsupportedSourceError + client = _mock_client() - client.create_document_from_source.side_effect = ValueError( + client.create_document_from_source.side_effect = UnsupportedSourceError( "Invalid S3 URI: s3:///bad" ) with pytest.raises(PermanentError): @@ -135,8 +141,10 @@ async def test_invalid_s3_uri_classified_permanent(): @pytest.mark.asyncio async def test_missing_file_classified_permanent(): + from haiku.rag.client.exceptions import UnsupportedSourceError + client = _mock_client() - client.create_document_from_source.side_effect = ValueError( + client.create_document_from_source.side_effect = UnsupportedSourceError( "File does not exist: /nope" ) with pytest.raises(PermanentError):