diff --git a/src/openai/_base_client.py b/src/openai/_base_client.py index 7f92a61a86..3c4f6d1f0b 100644 --- a/src/openai/_base_client.py +++ b/src/openai/_base_client.py @@ -591,7 +591,8 @@ def _build_request( elif not files: # Don't set content when JSON is sent as multipart/form-data, # since httpx's content param overrides other body arguments - kwargs["content"] = openapi_dumps(json_data) if is_given(json_data) and json_data is not None else None + if is_given(json_data) and json_data is not None: + kwargs["content"] = openapi_dumps(json_data) kwargs["files"] = files else: headers.pop("Content-Type", None) @@ -1617,6 +1618,114 @@ async def _send_request( ) -> httpx2.Response: return await self._client.send(request, stream=stream, **kwargs) + async def _build_request_async( + self, + options: FinalRequestOptions, + *, + retries_taken: int = 0, + ) -> httpx2.Request: + """Async-safe version of _build_request that runs JSON serialization in a thread pool.""" + # Request bodies, files, URLs, and custom options can contain private data. + log.debug( + "Building HTTP request: method=%s retries_taken=%i", + get_http_method_for_logging(options.method), + retries_taken, + ) + kwargs: dict[str, Any] = {} + + json_data = options.json_data + if options.extra_json is not None: + if json_data is None: + json_data = cast(Body, options.extra_json) + elif is_mapping(json_data): + json_data = _merge_mappings(json_data, options.extra_json) + else: + raise RuntimeError(f"Unexpected JSON data type, {type(json_data)}, cannot merge with `extra_body`") + + headers = self._build_headers(options, retries_taken=retries_taken) + params = _merge_mappings({**self._auth_query(options.security), **self.default_query}, options.params) + content_type = headers.get("Content-Type") + files = options.files + + # If the given Content-Type header is multipart/form-data then it + # has to be removed so that httpx can generate the header with + # additional information for us as it has to be in this form + # for the server to be able to correctly parse the request: + # multipart/form-data; boundary=---abc-- + if content_type is not None and content_type.startswith("multipart/form-data"): + if "boundary" not in content_type: + # only remove the header if the boundary hasn't been explicitly set + # as the caller doesn't want httpx to come up with their own boundary + headers.pop("Content-Type") + + # As we are now sending multipart/form-data instead of application/json + # we need to tell httpx to use it, https://www.python-httpx.org/advanced/clients/#multipart-file-encoding + if json_data: + if not is_dict(json_data): + raise TypeError( + f"Expected query input to be a dictionary for multipart requests but got {type(json_data)} instead." + ) + kwargs["data"] = self._serialize_multipartform(json_data) + + # httpx determines whether or not to send a "multipart/form-data" + # request based on the truthiness of the "files" argument. + # This gets around that issue by generating a dict value that + # evaluates to true. + # + # https://github.com/encode/httpx/discussions/2399#discussioncomment-3814186 + if not files: + files = cast(HttpxRequestFiles, ForceMultipartDict()) + + prepared_url = self._prepare_url(options.url) + # preserve hard-coded query params from the url + if params and prepared_url.query: + params = {**dict(prepared_url.params.items()), **params} + prepared_url = prepared_url.copy_with(raw_path=prepared_url.raw_path.split(b"?", 1)[0]) + + is_body_allowed = options.method.lower() != "get" + + if is_body_allowed: + if options.content is not None and json_data is not None: + raise TypeError("Passing both `content` and `json_data` is not supported") + if options.content is not None and files is not None: + raise TypeError("Passing both `content` and `files` is not supported") + if options.content is not None: + kwargs["content"] = options.content + elif isinstance(json_data, bytes): + kwargs["content"] = json_data + elif not files: + # Don't set content when JSON is sent as multipart/form-data, + # since httpx's content param overrides other body arguments + if is_given(json_data) and json_data is not None: + # Use async serialization to avoid blocking the event loop + kwargs["content"] = await asyncify(openapi_dumps)(json_data) + kwargs["files"] = files + else: + headers.pop("Content-Type", None) + kwargs.pop("data", None) + + timeout = self.timeout if isinstance(options.timeout, NotGiven) else options.timeout + request_url = str(prepared_url) + request_headers = list(headers.multi_items()) + if is_legacy_httpx_sync_client(self._client) or is_legacy_httpx_async_client(self._client): + timeout = normalize_legacy_httpx_timeout(timeout) + else: + timeout = normalize_httpx2_timeout(timeout) + + # TODO: report this error to httpx + return self._client.build_request( # pyright: ignore[reportUnknownMemberType] + headers=request_headers, + timeout=timeout, + method=options.method, + url=request_url, + # the `Query` type that we use is incompatible with qs' + # `Params` type as it needs to be typed as `Mapping[str, object]` + # so that passing a `TypedDict` doesn't cause an error. + # https://github.com/microsoft/pyright/issues/3526#event-6715453066 + params=self.qs.stringify(cast(Mapping[str, Any], params)) if params else None, + **kwargs, + ) + @overload async def request( self, @@ -1678,7 +1787,7 @@ async def request( options = await self._prepare_options(options) remaining_retries = max_retries - retries_taken - request = self._build_request(options, retries_taken=retries_taken) + request = await self._build_request_async(options, retries_taken=retries_taken) await self._prepare_request(request) kwargs: HttpxSendArgs = {} diff --git a/src/openai/lib/azure.py b/src/openai/lib/azure.py index c7e61767a2..671fa2c731 100644 --- a/src/openai/lib/azure.py +++ b/src/openai/lib/azure.py @@ -155,6 +155,24 @@ def _build_request( request.extensions[_AZURE_AUTH_ORIGIN] = _origin(request.url) return request + async def _build_request_async( + self, + options: FinalRequestOptions, + *, + retries_taken: int = 0, + ) -> httpx2.Request: + """Async variant of _build_request for use in async contexts.""" + if options.url in _deployments_endpoints and is_mapping(options.json_data): + model = options.json_data.get("model") + if model is not None and "/deployments" not in str(self.base_url.path): + options.url = path_template("/deployments/{model}", model=model) + options.url + + request = await super()._build_request_async(options, retries_taken=retries_taken) + # HTTPX preserves request extensions through redirects. Scope the hook + # to this Azure request, including when its HTTP client is shared. + request.extensions[_AZURE_AUTH_ORIGIN] = _origin(request.url) + return request + @override def _prepare_url(self, url: str) -> httpx2.URL: """Adjust the URL if the client was configured with an Azure endpoint + deployment diff --git a/tests/test_event_loop_blocking.py b/tests/test_event_loop_blocking.py new file mode 100644 index 0000000000..f0ea251288 --- /dev/null +++ b/tests/test_event_loop_blocking.py @@ -0,0 +1,105 @@ +"""Regression tests for event loop blocking during JSON serialization. + +This test verifies that async requests don't block the event loop during JSON +serialization, which is critical for concurrent operations like Redis, Kafka, +and WebSocket communication that share the event loop. + +See: https://github.com/openai/openai-python/issues/3777 +""" + +from __future__ import annotations + +import asyncio +from typing import AsyncIterator + +import httpx2 +import pytest + +from openai import AsyncOpenAI +from tests.respx2 import MockRouter + + +@pytest.mark.asyncio +async def test_async_request_does_not_block_event_loop( + respx2_mock: MockRouter, + async_client: AsyncOpenAI, +) -> None: + """Test that async JSON serialization doesn't block the event loop. + + This test verifies the fix for issue #3777 by ensuring that: + 1. Background concurrent work completes while serialization happens + 2. The event loop remains responsive during JSON serialization + 3. Multiple concurrent tasks can progress simultaneously + """ + # Track when the background task completes + background_task_started = asyncio.Event() + background_task_done = asyncio.Event() + background_task_iterations = 0 + + async def background_work() -> None: + """Simulates concurrent work (e.g., Redis access, WebSocket read).""" + nonlocal background_task_iterations + background_task_started.set() + + # Run for a short time to give the event loop a chance to be blocked + for _ in range(100): + background_task_iterations += 1 + await asyncio.sleep(0.001) # 1ms per iteration = 100ms total + + background_task_done.set() + + # Setup the mock to return a successful response + respx2_mock.post("/chat/completions").mock( + return_value=httpx2.Response( + status_code=200, + json={ + "id": "chatcmpl-test", + "object": "chat.completion", + "created": 1234567890, + "model": "gpt-4", + "choices": [ + { + "index": 0, + "message": {"role": "assistant", "content": "Hello!"}, + "finish_reason": "stop", + } + ], + "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}, + }, + ) + ) + + # Start the background task + background_task = asyncio.create_task(background_work()) + + # Wait for background task to start + await asyncio.wait_for(background_task_started.wait(), timeout=1.0) + + # Make an async API request (which triggers JSON serialization) + # This should NOT block the event loop, allowing background_work to continue + response = await async_client.chat.completions.create( + model="gpt-4", + messages=[{"role": "user", "content": "Hello"}], + ) + + # Wait for background task to complete + await asyncio.wait_for(background_task_done.wait(), timeout=5.0) + await background_task + + # Verify the response was successful + assert response.id == "chatcmpl-test" + assert response.choices[0].message.content == "Hello!" + + # The critical assertion: background task must have made significant progress + # If the event loop was blocked during serialization, the background task + # would complete much later (only after the request completes). + # With proper async serialization, the background task should complete + # most of its iterations during the request. + # + # We expect at least 50 iterations out of 100 to have completed. + # This threshold allows for small timing variations while still catching + # any significant event loop blocking. + assert background_task_iterations >= 50, ( + f"Background task only completed {background_task_iterations}/100 iterations. " + f"Event loop may be getting blocked during JSON serialization." + )