Merge fe81b84b31 into eddfb5fb20

litellm/llms/custom_httpx/llm_http_handler.py

@@ -44,7 +44,7 @@ from litellm.llms.base_llm.base_model_iterator import (
     MockResponseIterator,
 )
 from litellm.llms.base_llm.batches.transformation import BaseBatchesConfig
-from litellm.llms.base_llm.chat.transformation import BaseConfig
+from litellm.llms.base_llm.chat.transformation import BaseConfig, BaseLLMException
 from litellm.llms.base_llm.containers.transformation import BaseContainerConfig
 from litellm.llms.base_llm.embedding.transformation import BaseEmbeddingConfig
 from litellm.llms.base_llm.evals.transformation import BaseEvalsAPIConfig
@@ -310,6 +310,17 @@ def _collect_ws_project_quota_callbacks() -> tuple[ProjectQuotaCallback, ...]:
     )
 
 
+def _find_base_llm_exception(error: BaseException) -> BaseLLMException | None:
+    current: BaseException | None = error
+    for _ in range(10):
+        if current is None:
+            return None
+        if isinstance(current, BaseLLMException):
+            return current
+        current = current.__cause__  # rebind-ok: traverse the bounded exception cause chain
+    return None
+
+
 class BaseLLMHTTPHandler:
     async def _make_common_async_call(
         self,
@@ -6037,6 +6048,10 @@ class BaseLLMHTTPHandler:
             BaseEvalsAPIConfig,
         ],
     ):
+        provider_exception: Final = _find_base_llm_exception(e)
+        if provider_exception is not None:
+            raise provider_exception
+
         received_status_code: Final = (
             e.response.status_code if isinstance(e, httpx.HTTPStatusError) else getattr(e, "status_code", None)
         )
@@ -6057,8 +6072,6 @@ class BaseLLMHTTPHandler:
             error_headers = {}
 
         if provider_config is None:
-            from litellm.llms.base_llm.chat.transformation import BaseLLMException
-
             raise BaseLLMException(
                 status_code=status_code,
                 message=error_text,

litellm/llms/vertex_ai/files/transformation.py

@@ -564,16 +564,15 @@ def _openai_batch_jsonl_entry_to_vertex_rows(
     return ({"request": vertex_request_body},)
 
 
-def _iter_stripped_lines(raw_lines: Iterable[str | bytes]) -> Iterator[str]:
+def _iter_numbered_stripped_lines(raw_lines: Iterable[str | bytes]) -> Iterator[tuple[int, str]]:
     """Decode (when needed), strip, and drop blank lines from an iterable of lines."""
-    for raw in raw_lines:
-        line = raw.decode("utf-8") if isinstance(raw, (bytes, bytearray)) else raw
-        line = line.strip()
+    for lineno, raw in enumerate(raw_lines, start=1):
+        line = (raw.decode("utf-8") if isinstance(raw, (bytes, bytearray)) else raw).strip()
         if line:
-            yield line
+            yield lineno, line
 
 
-def _iter_openai_jsonl_lines(openai_file_content: FileTypes) -> Iterator[str]:
+def _iter_numbered_openai_jsonl_lines(openai_file_content: FileTypes) -> Iterator[tuple[int, str]]:
     """
     Yield non-empty JSONL lines one at a time without materializing the whole
     payload, so peak memory stays bounded regardless of payload size. Mirrors
@@ -589,24 +588,26 @@ def _iter_openai_jsonl_lines(openai_file_content: FileTypes) -> Iterator[str]:
         # into a BytesIO just to iterate it line by line.
         newline: Final = ord("\n")
         start, length = 0, len(content)
+        lineno: int = 0  # rebind-ok: line counter advances for each streamed chunk
         while start < length:
             idx = content.find(newline, start)
             if idx == -1:
                 chunk, start = content[start:], length
             else:
                 chunk, start = content[start:idx], idx + 1
+            lineno += 1
             line = chunk.decode("utf-8").strip()
             if line:
-                yield line
+                yield lineno, line
         return
 
     if isinstance(content, str):
-        yield from _iter_stripped_lines(io.StringIO(content))
+        yield from _iter_numbered_stripped_lines(io.StringIO(content))
         return
 
     if isinstance(content, PathLike):
         with open(str(content), "rb") as handle:
-            yield from _iter_stripped_lines(handle)
+            yield from _iter_numbered_stripped_lines(handle)
         return
 
     if hasattr(content, "read"):
@@ -627,17 +628,28 @@ def _iter_openai_jsonl_lines(openai_file_content: FileTypes) -> Iterator[str]:
                 "Batch upload file handle must be seekable so it can be re-read "
                 "for the GCS object name and the upload body."
             ) from e
-        yield from _iter_stripped_lines(content)
+        yield from _iter_numbered_stripped_lines(content)
         return
 
     raise ValueError("Unsupported file content type")
 
 
+def _iter_openai_jsonl_lines(openai_file_content: FileTypes) -> Iterator[str]:
+    yield from (line for _, line in _iter_numbered_openai_jsonl_lines(openai_file_content))
+
+
 def _iter_openai_jsonl_entries(
     openai_file_content: FileTypes,
 ) -> Iterator[dict[str, Any]]:
-    for line in _iter_openai_jsonl_lines(openai_file_content):
-        yield json.loads(line)
+    for lineno, line in _iter_numbered_openai_jsonl_lines(openai_file_content):
+        try:
+            entry: dict[str, Any] = json.loads(line)
+        except json.JSONDecodeError as e:
+            raise VertexAIError(
+                status_code=400,
+                message=f"Invalid JSON on line {lineno} of batch input file: {e.msg}",
+            ) from e
+        yield entry
 
 
 def _parse_vertex_batch_output_row(line: str) -> _VertexBatchRow:

tests/test_litellm/llms/custom_httpx/test_llm_http_handler.py

@@ -2334,6 +2334,47 @@ def test_sync_retrieve_file_content_raises_on_http_error():
     assert exc_info.value.status_code == 404
 
 
+def test_handle_error_preserves_base_llm_exception_message_and_status():
+    from litellm.llms.vertex_ai.common_utils import VertexAIError
+    from litellm.llms.vertex_ai.files.transformation import VertexAIFilesConfig
+
+    message = "Invalid JSON on line 3 of batch input file: Expecting value"
+    error = VertexAIError(status_code=400, message=message)
+
+    with pytest.raises(VertexAIError) as exc_info:
+        BaseLLMHTTPHandler()._handle_error(e=error, provider_config=VertexAIFilesConfig())
+
+    assert exc_info.value.status_code == 400
+    assert exc_info.value.message == message
+
+
+def test_handle_error_preserves_base_llm_exception_from_wrapper():
+    from litellm.llms.vertex_ai.common_utils import VertexAIError
+    from litellm.llms.vertex_ai.files.transformation import VertexAIFilesConfig
+
+    message = "Invalid JSON on line 3 of batch input file: Expecting value"
+    error = VertexAIError(status_code=400, message=message)
+    wrapper = httpx.ConnectError("Failed to send bytes")
+    wrapper.__cause__ = error
+
+    with pytest.raises(VertexAIError) as exc_info:
+        BaseLLMHTTPHandler()._handle_error(e=wrapper, provider_config=VertexAIFilesConfig())
+
+    assert exc_info.value.status_code == 400
+    assert exc_info.value.message == message
+
+
+def test_handle_error_maps_plain_exception_to_provider_error():
+    from litellm.llms.vertex_ai.common_utils import VertexAIError
+    from litellm.llms.vertex_ai.files.transformation import VertexAIFilesConfig
+
+    with pytest.raises(VertexAIError) as exc_info:
+        BaseLLMHTTPHandler()._handle_error(e=ValueError("boom"), provider_config=VertexAIFilesConfig())
+
+    assert exc_info.value.status_code == 500
+    assert exc_info.value.message == "boom"
+
+
 _UPSTREAM_NOT_FOUND_BODY = {
     "error": {
         "message": "Response with id 'resp_abc' not found.",

tests/test_litellm/llms/vertex_ai/files/test_vertex_ai_files_streaming.py

@@ -31,16 +31,16 @@ from litellm.litellm_core_utils.litellm_logging import Logging
 from litellm.llms.base_llm.files.transformation import BaseFileUploadStream
 from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
 from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
+from litellm.llms.vertex_ai.common_utils import VertexAIError
 from litellm.llms.vertex_ai.files.transformation import (
     VertexAIFilesConfig,
-    _OpenAIToVertexBatchUploadStream,
     _get_litellm_batch_custom_id_from_labels,
     _iter_openai_jsonl_entries,
     _iter_openai_jsonl_lines,
     _openai_batch_jsonl_entry_to_vertex_rows,
+    _OpenAIToVertexBatchUploadStream,
 )
 from litellm.types.llms.openai import CreateFileRequest
-from litellm.llms.vertex_ai.common_utils import VertexAIError
 
 
 def _upload_stream(transformed) -> BaseFileUploadStream:
@@ -221,8 +221,10 @@ class TestStreamingLineIterator:
         content = b'{"custom_id": "first"}\nnot-json-at-all\n'
         gen = _iter_openai_jsonl_entries(content)
         assert next(gen)["custom_id"] == "first"
-        with pytest.raises(json.JSONDecodeError):
+        with pytest.raises(VertexAIError) as exc_info:
             next(gen)
+        assert exc_info.value.status_code == 400
+        assert "line 2" in str(exc_info.value)
 
 
 class TestGetObjectNameLazyParse:
@@ -476,6 +478,63 @@ class TestUploadUrl:
 
 
 class TestUploadStreamBody:
+    def test_stream_rejects_malformed_single_row(self):
+        stream = _OpenAIToVertexBatchUploadStream(
+            openai_file_content=b"this is not json\n",
+            map_openai_to_vertex_params=lambda body: body,
+        )
+        with pytest.raises(VertexAIError) as exc_info:
+            list(stream.iter_bytes())
+        assert exc_info.value.status_code == 400
+        assert "line 1" in str(exc_info.value)
+
+    def test_stream_reports_file_line_number_for_malformed_file_like_input(self):
+        valid_row = json.dumps(
+            {
+                "custom_id": "r1",
+                "method": "POST",
+                "url": "/v1/chat/completions",
+                "body": {
+                    "model": "gemini-2.5-flash",
+                    "messages": [{"role": "user", "content": "hi"}],
+                },
+            }
+        ).encode("utf-8")
+        raw = valid_row + b"\n\nthis is not json\n"
+        stream = _OpenAIToVertexBatchUploadStream(
+            openai_file_content=io.BytesIO(raw),
+            map_openai_to_vertex_params=lambda body: body,
+        )
+        with pytest.raises(VertexAIError) as exc_info:
+            list(stream.iter_bytes())
+        assert "line 3" in str(exc_info.value)
+
+    def test_stream_reports_file_line_number_for_malformed_bytes_input(self):
+        valid_row = json.dumps(
+            {
+                "custom_id": "r1",
+                "method": "POST",
+                "url": "/v1/chat/completions",
+                "body": {
+                    "model": "gemini-2.5-flash",
+                    "messages": [{"role": "user", "content": "hi"}],
+                },
+            }
+        ).encode("utf-8")
+        raw = valid_row + b"\n\nthis is not json\n"
+        stream = _OpenAIToVertexBatchUploadStream(
+            openai_file_content=raw,
+            map_openai_to_vertex_params=lambda body: body,
+        )
+        with pytest.raises(VertexAIError) as exc_info:
+            list(stream.iter_bytes())
+        assert "line 3" in str(exc_info.value)
+
+    def test_get_object_name_rejects_malformed_batch_jsonl(self):
+        with pytest.raises(VertexAIError) as exc_info:
+            VertexAIFilesConfig().get_object_name(file_data=b"not json\n", purpose="batch")
+        assert exc_info.value.status_code == 400
+
     def test_stream_matches_legacy_pipeline(self):
         cfg = VertexAIFilesConfig()
         raw = _make_openai_jsonl_bytes(120)