From b491f767f3a7eac34dd96fe09d27923e45c2c392 Mon Sep 17 00:00:00 2001 From: Yiorgis Gozadinos Date: Tue, 26 May 2026 17:38:49 +0300 Subject: [PATCH] Let httpx errors flow through DoclingServeClient --- .../haiku/rag/ingester/workers/pipeline.py | 7 +- .../haiku/rag/providers/docling_serve.py | 92 +++++-------------- tests/ingester/test_pipeline.py | 21 +++++ tests/test_chunker.py | 11 ++- tests/test_converters.py | 11 ++- 5 files changed, 59 insertions(+), 83 deletions(-) diff --git a/haiku_rag_slim/haiku/rag/ingester/workers/pipeline.py b/haiku_rag_slim/haiku/rag/ingester/workers/pipeline.py index c4b566d2..45b9babf 100644 --- a/haiku_rag_slim/haiku/rag/ingester/workers/pipeline.py +++ b/haiku_rag_slim/haiku/rag/ingester/workers/pipeline.py @@ -52,10 +52,9 @@ def _classify(exc: BaseException) -> Exception: return TransientError(f"HTTP {status}: {exc}") return PermanentError(f"HTTP {status}: {exc}") - if isinstance( - exc, - httpx.ConnectError | httpx.ReadTimeout | httpx.WriteTimeout | httpx.PoolTimeout, - ): + if isinstance(exc, httpx.TransportError): + # Umbrella for ConnectError, NetworkError, TimeoutException, ProtocolError, + # ProxyError — every transport-layer failure that's worth retrying. return TransientError(f"network: {exc}") if isinstance(exc, asyncio.TimeoutError | TimeoutError | OSError): diff --git a/haiku_rag_slim/haiku/rag/providers/docling_serve.py b/haiku_rag_slim/haiku/rag/providers/docling_serve.py index 9d7b7930..2d32e98c 100644 --- a/haiku_rag_slim/haiku/rag/providers/docling_serve.py +++ b/haiku_rag_slim/haiku/rag/providers/docling_serve.py @@ -118,50 +118,22 @@ class DoclingServeClient: ) -> dict[str, Any]: """Submit a task and poll until completion; fetch result as JSON. - Args: - endpoint: The async endpoint path (e.g., "/v1/convert/file/async") - files: Files to upload - data: Form data parameters - name: Name for error messages - - Returns: - The result dictionary from the completed task - - Raises: - ValueError: If the task fails or service is unavailable + httpx exceptions (ConnectError, HTTPStatusError, TimeoutException, + etc.) propagate so the ingester's pipeline classifier can route + 4xx → PermanentError and 5xx/network → TransientError. ValueError + is raised by `_submit_and_wait` when docling-serve reports a task + failure or returns no task_id. """ headers = self._get_headers() base_url = self._pick_url() - - try: - async with self._httpx_client() as client: - task_id = await self._submit_and_wait( - client, base_url, endpoint, files, data, headers, name - ) - result_url = f"{base_url}/v1/result/{task_id}" - result_response = await client.get(result_url, headers=headers) - result_response.raise_for_status() - return result_response.json() - - except httpx.ConnectError as e: - raise ValueError( - f"Could not connect to docling-serve at {base_url}. " - f"Ensure the service is running and accessible. Error: {e}" - ) from e - except httpx.TimeoutException as e: - raise ValueError( - f"Request to docling-serve timed out after {self.timeout}s. Error: {e}" - ) from e - except httpx.HTTPStatusError as e: - if e.response.status_code == 401: - raise ValueError( - "Authentication failed. Check your API key configuration." - ) from e - raise ValueError(f"HTTP error from docling-serve: {e}") from e - except ValueError: - raise - except Exception as e: - raise ValueError(f"Failed to process via docling-serve: {e}") from e + async with self._httpx_client() as client: + task_id = await self._submit_and_wait( + client, base_url, endpoint, files, data, headers, name + ) + result_url = f"{base_url}/v1/result/{task_id}" + result_response = await client.get(result_url, headers=headers) + result_response.raise_for_status() + return result_response.json() async def submit_and_poll_zip( self, @@ -179,33 +151,11 @@ class DoclingServeClient: """ headers = self._get_headers() base_url = self._pick_url() - - try: - async with self._httpx_client() as client: - task_id = await self._submit_and_wait( - client, base_url, endpoint, files, data, headers, name - ) - result_url = f"{base_url}/v1/result/{task_id}" - result_response = await client.get(result_url, headers=headers) - result_response.raise_for_status() - return result_response.content - - except httpx.ConnectError as e: - raise ValueError( - f"Could not connect to docling-serve at {base_url}. " - f"Ensure the service is running and accessible. Error: {e}" - ) from e - except httpx.TimeoutException as e: - raise ValueError( - f"Request to docling-serve timed out after {self.timeout}s. Error: {e}" - ) from e - except httpx.HTTPStatusError as e: - if e.response.status_code == 401: - raise ValueError( - "Authentication failed. Check your API key configuration." - ) from e - raise ValueError(f"HTTP error from docling-serve: {e}") from e - except ValueError: - raise - except Exception as e: - raise ValueError(f"Failed to process via docling-serve: {e}") from e + async with self._httpx_client() as client: + task_id = await self._submit_and_wait( + client, base_url, endpoint, files, data, headers, name + ) + result_url = f"{base_url}/v1/result/{task_id}" + result_response = await client.get(result_url, headers=headers) + result_response.raise_for_status() + return result_response.content diff --git a/tests/ingester/test_pipeline.py b/tests/ingester/test_pipeline.py index be16c61a..0ac1f0d1 100644 --- a/tests/ingester/test_pipeline.py +++ b/tests/ingester/test_pipeline.py @@ -196,6 +196,27 @@ async def test_connect_error_classified_transient(): await run_job(client, _job()) +@pytest.mark.parametrize( + "exc_factory", + [ + lambda: httpx.ConnectTimeout("slow connect"), + lambda: httpx.ReadTimeout("slow read"), + lambda: httpx.WriteTimeout("slow write"), + lambda: httpx.PoolTimeout("pool"), + lambda: httpx.ProxyError("proxy"), + ], +) +@pytest.mark.asyncio +async def test_transport_subclasses_classified_transient(exc_factory): + """The classifier umbrellas on httpx.TransportError so every transport- + layer subclass routes to TransientError, not the generic 'unexpected' + fallback.""" + client = _mock_client() + client.create_document_from_source.side_effect = exc_factory() + with pytest.raises(TransientError, match="network"): + await run_job(client, _job()) + + @pytest.mark.asyncio async def test_unknown_exception_classified_transient(): client = _mock_client() diff --git a/tests/test_chunker.py b/tests/test_chunker.py index f09f600d..5f9f78b8 100644 --- a/tests/test_chunker.py +++ b/tests/test_chunker.py @@ -443,7 +443,7 @@ class TestDoclingServeChunker: converter = get_converter(Config) doc = await converter.convert_text("# Test", name="test.md") - with pytest.raises(ValueError, match="Could not connect to docling-serve"): + with pytest.raises(httpx.ConnectError): await chunker.chunk(doc) @pytest.mark.asyncio @@ -459,13 +459,15 @@ class TestDoclingServeChunker: converter = get_converter(Config) doc = await converter.convert_text("# Test", name="test.md") - with pytest.raises(ValueError, match="timed out"): + with pytest.raises(httpx.TimeoutException): await chunker.chunk(doc) @pytest.mark.asyncio @patch("haiku.rag.providers.docling_serve.httpx.AsyncClient") async def test_chunk_auth_error(self, mock_client_class, chunker): - """Test handling of authentication errors.""" + """Auth failures surface as httpx.HTTPStatusError(401) so the + ingester's pipeline classifier can route them to PermanentError — + retrying a bad token is wasted work.""" import httpx mock_request = Mock() @@ -482,8 +484,9 @@ class TestDoclingServeChunker: converter = get_converter(Config) doc = await converter.convert_text("# Test", name="test.md") - with pytest.raises(ValueError, match="Authentication failed"): + with pytest.raises(httpx.HTTPStatusError) as exc_info: await chunker.chunk(doc) + assert exc_info.value.response.status_code == 401 @pytest.mark.asyncio @patch("haiku.rag.providers.docling_serve.httpx.AsyncClient") diff --git a/tests/test_converters.py b/tests/test_converters.py index 8514e713..aeb64f93 100644 --- a/tests/test_converters.py +++ b/tests/test_converters.py @@ -1130,7 +1130,7 @@ class TestDoclingServeConverter: mock_client.__aexit__ = AsyncMock(return_value=None) mock_client_class.return_value = mock_client - with pytest.raises(ValueError, match="Could not connect to docling-serve"): + with pytest.raises(httpx.ConnectError): await converter.convert_text("# Test") @pytest.mark.asyncio @@ -1143,12 +1143,14 @@ class TestDoclingServeConverter: mock_client.__aexit__ = AsyncMock(return_value=None) mock_client_class.return_value = mock_client - with pytest.raises(ValueError, match="timed out"): + with pytest.raises(httpx.TimeoutException): await converter.convert_text("# Test") @pytest.mark.asyncio async def test_convert_text_auth_error(self, converter): - """Test handling of authentication errors.""" + """Auth failures surface as httpx.HTTPStatusError(401) so the + ingester's pipeline classifier can route them to PermanentError — + retrying a bad token is wasted work.""" mock_response = Mock() mock_response.status_code = 401 @@ -1163,8 +1165,9 @@ class TestDoclingServeConverter: mock_client.__aexit__ = AsyncMock(return_value=None) mock_client_class.return_value = mock_client - with pytest.raises(ValueError, match="Authentication failed"): + with pytest.raises(httpx.HTTPStatusError) as exc_info: await converter.convert_text("# Test") + assert exc_info.value.response.status_code == 401 @pytest.mark.asyncio async def test_convert_file_pdf(self, converter):