fix(passthrough): extract _build_passthrough_logging_result helper

The #29311 cherry-pick onto stable/1.85.x carried the test test_route_streaming_logging_runs_async_handler_for_sdk_passthrough, which patches PassThroughStreamingHandler._build_passthrough_logging_result to verify the SDK-passthrough dispatch contract. The manual conflict resolution kept the per-endpoint if/elif/elif chain inline in _route_streaming_logging_to_handler (matching v1.84.4's resolution), so the patched attribute did not exist and the test errored at collection with AttributeError. Extract the chain into the static _build_passthrough_logging_result helper as #29089 originally designed it. _route_streaming_logging_to_handler now resolves (standard_logging_response_object, kwargs) through the helper and dispatches via dispatch_success_handlers; the helper itself is synchronous and CPU-bound, suitable for the unit test's patch target. Verified locally: tests/pass_through_unit_tests/test_unit_test_streaming.py passes (5/5) and tests/test_litellm/litellm_core_utils/test_litellm_logging.py passes (83/83). v1.84.4 ships with the same broken test; this strictly improves on that resolution.
2026-08-24 00:28:32 +00:00 · 2026-06-01 17:31:05 -07:00
parent acbbfe9cae
commit 8824745c4e
1 changed files with 97 additions and 58 deletions
@@ -1,6 +1,6 @@
 import asyncio
 from datetime import datetime
-from typing import List, Optional
+from typing import List, Optional, Tuple

 import httpx

@@ -114,64 +114,20 @@ class PassThroughStreamingHandler:
        - OpenAI
        """
        try:
-            all_chunks = PassThroughStreamingHandler._convert_raw_bytes_to_str_lines(
-                raw_bytes
+            (
+                standard_logging_response_object,
+                kwargs,
+            ) = PassThroughStreamingHandler._build_passthrough_logging_result(
+                litellm_logging_obj=litellm_logging_obj,
+                passthrough_success_handler_obj=passthrough_success_handler_obj,
+                url_route=url_route,
+                request_body=request_body,
+                endpoint_type=endpoint_type,
+                start_time=start_time,
+                raw_bytes=raw_bytes,
+                end_time=end_time,
+                model=model,
            )
-            standard_logging_response_object: Optional[
-                PassThroughEndpointLoggingResultValues
-            ] = None
-            kwargs: dict = {}
-            if endpoint_type == EndpointType.ANTHROPIC:
-                anthropic_passthrough_logging_handler_result = AnthropicPassthroughLoggingHandler._handle_logging_anthropic_collected_chunks(
-                    litellm_logging_obj=litellm_logging_obj,
-                    passthrough_success_handler_obj=passthrough_success_handler_obj,
-                    url_route=url_route,
-                    request_body=request_body,
-                    endpoint_type=endpoint_type,
-                    start_time=start_time,
-                    all_chunks=all_chunks,
-                    end_time=end_time,
-                )
-                standard_logging_response_object = (
-                    anthropic_passthrough_logging_handler_result["result"]
-                )
-                kwargs = anthropic_passthrough_logging_handler_result["kwargs"]
-            elif endpoint_type == EndpointType.VERTEX_AI:
-                vertex_passthrough_logging_handler_result = VertexPassthroughLoggingHandler._handle_logging_vertex_collected_chunks(
-                    litellm_logging_obj=litellm_logging_obj,
-                    passthrough_success_handler_obj=passthrough_success_handler_obj,
-                    url_route=url_route,
-                    request_body=request_body,
-                    endpoint_type=endpoint_type,
-                    start_time=start_time,
-                    all_chunks=all_chunks,
-                    end_time=end_time,
-                    model=model,
-                )
-                standard_logging_response_object = (
-                    vertex_passthrough_logging_handler_result["result"]
-                )
-                kwargs = vertex_passthrough_logging_handler_result["kwargs"]
-            elif endpoint_type == EndpointType.OPENAI:
-                openai_passthrough_logging_handler_result = OpenAIPassthroughLoggingHandler._handle_logging_openai_collected_chunks(
-                    litellm_logging_obj=litellm_logging_obj,
-                    passthrough_success_handler_obj=passthrough_success_handler_obj,
-                    url_route=url_route,
-                    request_body=request_body,
-                    endpoint_type=endpoint_type,
-                    start_time=start_time,
-                    all_chunks=all_chunks,
-                    end_time=end_time,
-                )
-                standard_logging_response_object = (
-                    openai_passthrough_logging_handler_result["result"]
-                )
-                kwargs = openai_passthrough_logging_handler_result["kwargs"]
-
-            if standard_logging_response_object is None:
-                standard_logging_response_object = StandardPassThroughResponseObject(
-                    response=f"cannot parse chunks to standard response object. Chunks={all_chunks}"
-                )
            # Always reached from an async context (anthropic_messages,
            # google_genai, and proxy pass-through stream tasks). prefer_async_handlers
            # keeps async-only loggers running even when call_type isn't pass_through
@@ -189,6 +145,89 @@ class PassThroughStreamingHandler:
                f"Error in _route_streaming_logging_to_handler: {str(e)}"
            )

+    @staticmethod
+    def _build_passthrough_logging_result(
+        litellm_logging_obj: LiteLLMLoggingObj,
+        passthrough_success_handler_obj: PassThroughEndpointLogging,
+        url_route: str,
+        request_body: dict,
+        endpoint_type: EndpointType,
+        start_time: datetime,
+        raw_bytes: List[bytes],
+        end_time: datetime,
+        model: Optional[str],
+    ) -> Tuple[PassThroughEndpointLoggingResultValues, dict]:
+        """
+        Synchronous, CPU-bound reconstruction of the standard logging payload
+        from collected raw SSE bytes. Extracted from
+        _route_streaming_logging_to_handler so the per-endpoint dispatch can
+        be unit-tested in isolation. Still invoked synchronously on the event
+        loop; an off-loop dispatch is a future change, not part of this PR.
+        """
+        all_chunks = PassThroughStreamingHandler._convert_raw_bytes_to_str_lines(
+            raw_bytes
+        )
+        standard_logging_response_object: Optional[
+            PassThroughEndpointLoggingResultValues
+        ] = None
+        kwargs: dict = {}
+        if endpoint_type == EndpointType.ANTHROPIC:
+            anthropic_passthrough_logging_handler_result = AnthropicPassthroughLoggingHandler._handle_logging_anthropic_collected_chunks(
+                litellm_logging_obj=litellm_logging_obj,
+                passthrough_success_handler_obj=passthrough_success_handler_obj,
+                url_route=url_route,
+                request_body=request_body,
+                endpoint_type=endpoint_type,
+                start_time=start_time,
+                all_chunks=all_chunks,
+                end_time=end_time,
+            )
+            standard_logging_response_object = (
+                anthropic_passthrough_logging_handler_result["result"]
+            )
+            kwargs = anthropic_passthrough_logging_handler_result["kwargs"]
+        elif endpoint_type == EndpointType.VERTEX_AI:
+            vertex_passthrough_logging_handler_result = (
+                VertexPassthroughLoggingHandler._handle_logging_vertex_collected_chunks(
+                    litellm_logging_obj=litellm_logging_obj,
+                    passthrough_success_handler_obj=passthrough_success_handler_obj,
+                    url_route=url_route,
+                    request_body=request_body,
+                    endpoint_type=endpoint_type,
+                    start_time=start_time,
+                    all_chunks=all_chunks,
+                    end_time=end_time,
+                    model=model,
+                )
+            )
+            standard_logging_response_object = (
+                vertex_passthrough_logging_handler_result["result"]
+            )
+            kwargs = vertex_passthrough_logging_handler_result["kwargs"]
+        elif endpoint_type == EndpointType.OPENAI:
+            openai_passthrough_logging_handler_result = (
+                OpenAIPassthroughLoggingHandler._handle_logging_openai_collected_chunks(
+                    litellm_logging_obj=litellm_logging_obj,
+                    passthrough_success_handler_obj=passthrough_success_handler_obj,
+                    url_route=url_route,
+                    request_body=request_body,
+                    endpoint_type=endpoint_type,
+                    start_time=start_time,
+                    all_chunks=all_chunks,
+                    end_time=end_time,
+                )
+            )
+            standard_logging_response_object = (
+                openai_passthrough_logging_handler_result["result"]
+            )
+            kwargs = openai_passthrough_logging_handler_result["kwargs"]
+
+        if standard_logging_response_object is None:
+            standard_logging_response_object = StandardPassThroughResponseObject(
+                response=f"cannot parse chunks to standard response object. Chunks={all_chunks}"
+            )
+        return standard_logging_response_object, kwargs
+
    @staticmethod
    def _extract_model_for_cost_injection(
        request_body: Optional[dict],