From 35a224c58d1ea2a9af31336061c846b9ead5c938 Mon Sep 17 00:00:00 2001 From: markstuart-oai <323302876+markstuart-oai@users.noreply.github.com> Date: Wed, 21 Oct 2026 20:54:55 +0000 Subject: [PATCH] fix(api): honor per-call WebSocket query settings Castiron-Internal-PR: https://github.com/openai/openai-python-internal/pull/303 Castiron-Source-SHA: df11e60f60ebe9acf1cab6c741b66322b11e53b0 Castiron-Public-Base-SHA: 26e5d926966fc5c7e688790b21d66dc08d35eceb --- .castiron.stats.yml | 8 +- scripts/castiron/generated-files.json | 32 +-- src/openai/_base_client.py | 20 ++ src/openai/lib/azure.py | 12 +- src/openai/lib/realtime_translation.py | 21 +- .../resources/beta/realtime/realtime.py | 24 +- .../resources/beta/responses/responses.py | 14 +- src/openai/resources/live/forks.py | 14 +- src/openai/resources/live/live.py | 14 +- src/openai/resources/live/sideband.py | 32 +-- src/openai/resources/realtime/realtime.py | 32 +-- .../realtime/translations/translations.py | 28 +- src/openai/resources/responses/responses.py | 14 +- .../lib/test_websocket_query_serialization.py | 265 ++++++++++++++++++ 14 files changed, 403 insertions(+), 127 deletions(-) create mode 100644 tests/lib/test_websocket_query_serialization.py diff --git a/.castiron.stats.yml b/.castiron.stats.yml index 4048b88c15..bab06c97f5 100644 --- a/.castiron.stats.yml +++ b/.castiron.stats.yml @@ -1,8 +1,8 @@ schema_version: 1 -generation_id: d25554dd-0197-4996-acdc-7aade76e2045 +generation_id: cec0794a-44bb-431c-b91c-32bbcf798980 openapi_spec_hash: a7dcb5a78f21ffdb756ac4bbe0b8a227 openapi_transformed_spec_hash: da855097c89cd19cdb576823a15b0481 config_hash: a9d68fc59815e9884d653f40d0c9cc15 -codegen_sha: 2fd3ddb168b20dbf959dc9733c389b1be4d3a027 -codegen_hash: 7c1bfd93466328a0bae3a67e40c27251bc2152767cd7cf0c12835d7fac647efd -public_codegen_sha: efa5c8db4dbf18180716aff72b9a58d1a9d6d2d6 +codegen_sha: 82f70fc38cea881d2702056442d49bd401e6a181 +codegen_hash: 9faf3e77625fc5caccdf324dfd3f099f54178b901026cca8d6e1a558d16b19ff +public_codegen_sha: d0df48b54e26bd50035e182ba2c03a990d728f9e diff --git a/scripts/castiron/generated-files.json b/scripts/castiron/generated-files.json index 8e25542838..c2fd3d5680 100644 --- a/scripts/castiron/generated-files.json +++ b/scripts/castiron/generated-files.json @@ -3,11 +3,11 @@ "files": { ".github/workflows/castiron-custom-code-comment.yml": { "mode": "100644", - "sha256": "c51faae4a106b873852075c369847719f3f99e59fa3e1cf32ce02f8f9d697afc" + "sha256": "02b1c497461fcaf604e1389e7dcab06cfcca01635a8cb2e92f8769245a2aaead" }, ".github/workflows/castiron-custom-code.yml": { "mode": "100644", - "sha256": "ba6d4c8bcd35481bd88245f60a566b223dc5272fc1bcc1e6834e4a8db9c52baf" + "sha256": "f27d44cbe8caa54c1a41af4c55b2b909ae465a12982cb5a32d49deaa44ef3912" }, "api.md": { "mode": "100644", @@ -23,7 +23,7 @@ }, "scripts/castiron/custom_code_budget.py": { "mode": "100755", - "sha256": "2beaac2eb0a963e717bf2bba93f52d6aea28fa7c4961591d4c2c4749945dbb7c" + "sha256": "31b47b2c85fc7ef483c18a94f0725e4f6bae50164c427330af6d081d5d3f31f2" }, "scripts/castiron/custom_code_report.py": { "mode": "100755", @@ -35,7 +35,7 @@ }, "scripts/castiron/fixtures/github_publisher.cjs": { "mode": "100644", - "sha256": "114a35f8ed075f20d7fbcfab6dbbf16383866497a9d6e8290cf0928bba753468" + "sha256": "fc80b1aff054f8a2ea16fc6bd9cff5ed4e4a48eebc7f3af8427947796fb8d865" }, "scripts/castiron/ruff.toml": { "mode": "100644", @@ -43,19 +43,19 @@ }, "scripts/castiron/test_custom_code_budget.py": { "mode": "100644", - "sha256": "f355e1cb0ed88aa5a344cd6c68388802f5ed814d638b500e271b008cf07e0106" + "sha256": "cf4938bca739d01f1db2a421ec97e4fe8b41b8bc2d2c634a01247e36cb464649" }, "scripts/castiron/test_custom_code_github.py": { "mode": "100644", - "sha256": "f3df0d260edd990532b32f15ed9cb4071834634bbb08a87d0077481e09128f98" + "sha256": "aa170458b1b58523df78073a96ba8c58b75fce03206ff7d62d99b29791c6ca39" }, "scripts/castiron/test_custom_code_publication.py": { "mode": "100644", - "sha256": "1d93757664a6ffd26f21df2a4c07ac635393f5da1ab0283df75bad66faa2ef2e" + "sha256": "99b266761f427d77990e68ce2fdea4b5f1a5b175612b3a9f4355c094939130ca" }, "scripts/castiron/test_custom_code_report.py": { "mode": "100644", - "sha256": "4429459798fd4cd57f6bbc6c0a9bbe88d76304cba07d2d2d4bfc22b5d403bdda" + "sha256": "4a084ae9fb53897fa260b22baae4ef8a0662ff5f92b5efaaf49f3ec8953949bd" }, "src/openai/__init__.py": { "mode": "100644", @@ -415,7 +415,7 @@ }, "src/openai/resources/beta/responses/responses.py": { "mode": "100644", - "sha256": "c7e705185228ea5a1662dc0fe569e0c16bb7eadf38beff925f6067643d98feb6" + "sha256": "06695dc41af2cd645f7c29921cbd6addef28c10dde118112bcca171a5a218275" }, "src/openai/resources/beta/threads/__init__.py": { "mode": "100644", @@ -595,11 +595,11 @@ }, "src/openai/resources/live/forks.py": { "mode": "100644", - "sha256": "ff834c298a4566869e81e769be92bcbed216510557ff4eb344fd840bab2f2cd5" + "sha256": "21884921a1b65b4c1fcf1a520b4e433c1548f433810caa0aaec6ae5130897389" }, "src/openai/resources/live/live.py": { "mode": "100644", - "sha256": "a0ae2ae3eaaeccf152edca7999656be8cf10e5da49fc6335d479d71ba259b30d" + "sha256": "ec86c0a3868d736b4ab8586ea60bd736c14adaadfcd1d198099f3ca4229742f5" }, "src/openai/resources/live/sessions.py": { "mode": "100644", @@ -607,7 +607,7 @@ }, "src/openai/resources/live/sideband.py": { "mode": "100644", - "sha256": "37028100a5301b90b58cb6ed4ea2e20e0277fa476796d5d932cf848ad2a294c1" + "sha256": "7db3998d437169ca2437c8deb4cc047ee3148e6ec747d87d0b9c86ad2c9ee2cf" }, "src/openai/resources/models.py": { "mode": "100644", @@ -635,7 +635,7 @@ }, "src/openai/resources/realtime/realtime.py": { "mode": "100644", - "sha256": "29cf5fe34140d2f985151b264d3438897ed1579d7db1ef523a5831dd89ac7936" + "sha256": "acef2fd152ecca0c65ad82a42999ad2492d99cf49d2a91f8e905d61732d1560e" }, "src/openai/resources/realtime/translations/__init__.py": { "mode": "100644", @@ -647,7 +647,7 @@ }, "src/openai/resources/realtime/translations/translations.py": { "mode": "100644", - "sha256": "2ed29f4fa975a92d0067c7c28e30fefad8070fc4bd6a7b35ff124b0bbaeb94b5" + "sha256": "13f68331c7ac59780d6db2d4b5af79b92d6df43407fbc798add158a6d900d2ed" }, "src/openai/resources/responses/__init__.py": { "mode": "100644", @@ -667,7 +667,7 @@ }, "src/openai/resources/responses/responses.py": { "mode": "100644", - "sha256": "77b3bd59e28c76c8c54d21d556718d30d6be437f1b19507e40e3708a24986e77" + "sha256": "0b7c2ee504b7adb0f8bfbd16c91ee9de05a535f879708c57ffc33dde94ca4fa1" }, "src/openai/resources/safety/__init__.py": { "mode": "100644", @@ -7295,7 +7295,7 @@ }, "tests/api_resources/admin/organization/projects/test_api_keys.py": { "mode": "100644", - "sha256": "49c21791da11c49c01c26b40da8e826b37a496d8e209c4b76911be04435d5048" + "sha256": "1fa9769ccdc65d47c8c17c1e525a69e70ebb6d9248a017eb5fe4de7a337ec8ee" }, "tests/api_resources/admin/organization/projects/test_certificates.py": { "mode": "100644", diff --git a/src/openai/_base_client.py b/src/openai/_base_client.py index 8985106f4f..29bbfecc2d 100644 --- a/src/openai/_base_client.py +++ b/src/openai/_base_client.py @@ -59,6 +59,7 @@ AsyncBinaryTypes, HttpxRequestFiles, ModelBuilderProtocol, + omit, not_given, ) from ._utils import SensitiveHeadersFilter, is_dict, is_list, asyncify, is_given, lru_cache, is_mapping @@ -570,6 +571,25 @@ def _prepare_url(self, url: str) -> URL: return merge_url + def _websocket_query_params(self, url: URL, extra_query: Query) -> httpx2.QueryParams: + # Retire logical defaults before encoding so replacing a nested value + # cannot leave behind fields from the old value. + overridden = dict.fromkeys(extra_query, omit) + defaults = _merge_mappings(self.default_query, overridden) + params = httpx2.QueryParams(self.qs.stringify(cast(Any, defaults))).merge(url.params) + params = httpx2.QueryParams([(key, value) for key, value in params.multi_items() if key not in overridden]) + if isinstance(extra_query, httpx2.QueryParams): + return params.merge(extra_query) + # WebSocket per-call None and sequences historically used httpx's + # blank values and repeated keys, while defaults use the REST encoder. + values = _merge_mappings({}, extra_query) + for key, value in values.items(): + if value is None: + values[key] = "" + elif isinstance(value, (list, tuple)): + values[key] = ["" if item is None else item for item in cast(Iterable[object], value)] + return params.merge(self.qs.stringify(cast(Any, values), array_format="repeat")) + def _make_sse_decoder(self) -> SSEDecoder | SSEBytesDecoder: return SSEDecoder() diff --git a/src/openai/lib/azure.py b/src/openai/lib/azure.py index 996d201055..82416b0ca7 100644 --- a/src/openai/lib/azure.py +++ b/src/openai/lib/azure.py @@ -528,11 +528,7 @@ def _configure_realtime(self, model: str, extra_query: Query) -> tuple[httpx2.UR base_url = self._prepare_url("/realtime") realtime_url = base_url.copy_with(scheme="wss") - url = realtime_url.copy_with( - params=httpx2.QueryParams(self.qs.stringify(cast(Any, self.default_query))) - .merge(realtime_url.params) - .merge(cast(Any, query)) - ) + url = realtime_url.copy_with(params=self._websocket_query_params(realtime_url, query)) return url, auth_headers @@ -895,9 +891,5 @@ async def _configure_realtime(self, model: str, extra_query: Query) -> tuple[htt base_url = self._prepare_url("/realtime") realtime_url = base_url.copy_with(scheme="wss") - url = realtime_url.copy_with( - params=httpx2.QueryParams(self.qs.stringify(cast(Any, self.default_query))) - .merge(realtime_url.params) - .merge(cast(Any, query)) - ) + url = realtime_url.copy_with(params=self._websocket_query_params(realtime_url, query)) return url, auth_headers diff --git a/src/openai/lib/realtime_translation.py b/src/openai/lib/realtime_translation.py index 01b075ff33..39ae4bb906 100644 --- a/src/openai/lib/realtime_translation.py +++ b/src/openai/lib/realtime_translation.py @@ -49,12 +49,27 @@ def _check_connection(client: OpenAI | AsyncOpenAI, model: str, query: Query) -> raise ValueError("Realtime Translation does not accept an intent query parameter.") -def _connection_query(client: OpenAI | AsyncOpenAI, model: str, query: Query) -> Query: +class _PreparedTranslationQuery(httpx2.QueryParams): + """Complete managed query; its removed defaults must not be added again.""" + + +def _connection_query(client: OpenAI | AsyncOpenAI, model: str, query: Query) -> _PreparedTranslationQuery: # Preserve Translation's base URL query when a separate WebSocket gateway is used. - params = client.base_url.params.merge(client.qs.stringify(cast(Any, client.default_query))) + params = client.base_url.params + # Defaults override the HTTP base URL, while the WebSocket URL wins over both. + # Let the shared helper encode the defaults after retiring logical overrides. + default_params = httpx2.QueryParams(client.qs.stringify(cast(Any, client.default_query))) + params = httpx2.QueryParams([(key, value) for key, value in params.multi_items() if key not in default_params]) if client.websocket_base_url is not None: params = params.merge(httpx2.URL(str(client.websocket_base_url)).params) - return cast(Query, params.merge({"model": model}).merge(cast(Any, query))) + connection_query: Query = ( + httpx2.QueryParams({"model": model}).merge(query) + if isinstance(query, httpx2.QueryParams) + else {"model": model, **query} + ) + return _PreparedTranslationQuery( + client._websocket_query_params(client.base_url.copy_with(params=params), connection_query) + ) def _parse(data: str | bytes) -> RealtimeTranslationServerEvent: diff --git a/src/openai/resources/beta/realtime/realtime.py b/src/openai/resources/beta/realtime/realtime.py index 6f1f8b224f..f28086c468 100644 --- a/src/openai/resources/beta/realtime/realtime.py +++ b/src/openai/resources/beta/realtime/realtime.py @@ -366,17 +366,7 @@ async def __aenter__(self) -> AsyncRealtimeConnection: else: url = self._prepare_url() url = url.copy_with( - params=httpx2.QueryParams(self.__client.qs.stringify(cast(Any, self.__client.default_query))) - .merge(url.params) - .merge( - cast( - Any, - { - "model": self.__model, - **extra_query, - }, - ) - ), + params=self.__client._websocket_query_params(url, {"model": self.__model, **extra_query}), ) log.debug("Connecting to WebSocket API") if self.__websocket_connection_options: @@ -561,17 +551,7 @@ def __enter__(self) -> RealtimeConnection: else: url = self._prepare_url() url = url.copy_with( - params=httpx2.QueryParams(self.__client.qs.stringify(cast(Any, self.__client.default_query))) - .merge(url.params) - .merge( - cast( - Any, - { - "model": self.__model, - **extra_query, - }, - ) - ), + params=self.__client._websocket_query_params(url, {"model": self.__model, **extra_query}), ) log.debug("Connecting to WebSocket API") if self.__websocket_connection_options: diff --git a/src/openai/resources/beta/responses/responses.py b/src/openai/resources/beta/responses/responses.py index 6e61b51ebc..5bf7e019b0 100644 --- a/src/openai/resources/beta/responses/responses.py +++ b/src/openai/resources/beta/responses/responses.py @@ -4581,9 +4581,10 @@ async def _connect_ws(self, extra_query: Query, extra_headers: Headers) -> Async url = self._prepare_url() url = url.copy_with( - params=httpx2.QueryParams(self.__client.qs.stringify(cast(Any, self.__client.default_query))) - .merge(url.params) - .merge(cast(Any, extra_query)), + params=self.__client._websocket_query_params( + url, + extra_query, + ), ) url = url.copy_with(scheme={"http": "ws", "https": "wss"}.get(url.scheme, url.scheme)) options = await self.__client._prepare_options( @@ -4985,9 +4986,10 @@ def _connect_ws(self, extra_query: Query, extra_headers: Headers) -> WebSocketCo url = self._prepare_url() url = url.copy_with( - params=httpx2.QueryParams(self.__client.qs.stringify(cast(Any, self.__client.default_query))) - .merge(url.params) - .merge(cast(Any, extra_query)), + params=self.__client._websocket_query_params( + url, + extra_query, + ), ) url = url.copy_with(scheme={"http": "ws", "https": "wss"}.get(url.scheme, url.scheme)) options = self.__client._prepare_options( diff --git a/src/openai/resources/live/forks.py b/src/openai/resources/live/forks.py index 1cece08732..ae8ed5a0cc 100644 --- a/src/openai/resources/live/forks.py +++ b/src/openai/resources/live/forks.py @@ -459,9 +459,10 @@ async def _connect_ws(self, extra_query: Query, extra_headers: Headers) -> Async url = self._prepare_url() url = url.copy_with( - params=httpx2.QueryParams(self.__client.qs.stringify(cast(Any, self.__client.default_query))) - .merge(url.params) - .merge(cast(Any, extra_query)), + params=self.__client._websocket_query_params( + url, + extra_query, + ), ) url = url.copy_with(scheme={"http": "ws", "https": "wss"}.get(url.scheme, url.scheme)) options = await self.__client._prepare_options( @@ -869,9 +870,10 @@ def _connect_ws(self, extra_query: Query, extra_headers: Headers) -> WebSocketCo url = self._prepare_url() url = url.copy_with( - params=httpx2.QueryParams(self.__client.qs.stringify(cast(Any, self.__client.default_query))) - .merge(url.params) - .merge(cast(Any, extra_query)), + params=self.__client._websocket_query_params( + url, + extra_query, + ), ) url = url.copy_with(scheme={"http": "ws", "https": "wss"}.get(url.scheme, url.scheme)) options = self.__client._prepare_options( diff --git a/src/openai/resources/live/live.py b/src/openai/resources/live/live.py index 422057288e..82f6f66890 100644 --- a/src/openai/resources/live/live.py +++ b/src/openai/resources/live/live.py @@ -677,9 +677,10 @@ async def _connect_ws(self, extra_query: Query, extra_headers: Headers) -> Async url = self._prepare_url() url = url.copy_with( - params=httpx2.QueryParams(self.__client.qs.stringify(cast(Any, self.__client.default_query))) - .merge(url.params) - .merge(cast(Any, extra_query)), + params=self.__client._websocket_query_params( + url, + extra_query, + ), ) url = url.copy_with(scheme={"http": "ws", "https": "wss"}.get(url.scheme, url.scheme)) options = await self.__client._prepare_options( @@ -1083,9 +1084,10 @@ def _connect_ws(self, extra_query: Query, extra_headers: Headers) -> WebSocketCo url = self._prepare_url() url = url.copy_with( - params=httpx2.QueryParams(self.__client.qs.stringify(cast(Any, self.__client.default_query))) - .merge(url.params) - .merge(cast(Any, extra_query)), + params=self.__client._websocket_query_params( + url, + extra_query, + ), ) url = url.copy_with(scheme={"http": "ws", "https": "wss"}.get(url.scheme, url.scheme)) options = self.__client._prepare_options( diff --git a/src/openai/resources/live/sideband.py b/src/openai/resources/live/sideband.py index a3fb2ebc73..84c277dfcf 100644 --- a/src/openai/resources/live/sideband.py +++ b/src/openai/resources/live/sideband.py @@ -466,16 +466,12 @@ async def _connect_ws(self, extra_query: Query, extra_headers: Headers) -> Async url = self._prepare_url() url = url.copy_with( - params=httpx2.QueryParams(self.__client.qs.stringify(cast(Any, self.__client.default_query))) - .merge(url.params) - .merge( - cast( - Any, - { - **({"graceful_close": self.__graceful_close} if self.__graceful_close is not omit else {}), - **extra_query, - }, - ) + params=self.__client._websocket_query_params( + url, + { + **({"graceful_close": self.__graceful_close} if self.__graceful_close is not omit else {}), + **extra_query, + }, ), ) url = url.copy_with(scheme={"http": "ws", "https": "wss"}.get(url.scheme, url.scheme)) @@ -888,16 +884,12 @@ def _connect_ws(self, extra_query: Query, extra_headers: Headers) -> WebSocketCo url = self._prepare_url() url = url.copy_with( - params=httpx2.QueryParams(self.__client.qs.stringify(cast(Any, self.__client.default_query))) - .merge(url.params) - .merge( - cast( - Any, - { - **({"graceful_close": self.__graceful_close} if self.__graceful_close is not omit else {}), - **extra_query, - }, - ) + params=self.__client._websocket_query_params( + url, + { + **({"graceful_close": self.__graceful_close} if self.__graceful_close is not omit else {}), + **extra_query, + }, ), ) url = url.copy_with(scheme={"http": "ws", "https": "wss"}.get(url.scheme, url.scheme)) diff --git a/src/openai/resources/realtime/realtime.py b/src/openai/resources/realtime/realtime.py index 2ab8ced9f3..012f9f39c9 100644 --- a/src/openai/resources/realtime/realtime.py +++ b/src/openai/resources/realtime/realtime.py @@ -660,16 +660,12 @@ async def _connect_ws(self, extra_query: Query, extra_headers: Headers) -> Async else: url = self._prepare_url() url = url.copy_with( - params=httpx2.QueryParams(self.__client.qs.stringify(cast(Any, self.__client.default_query))) - .merge(url.params) - .merge( - cast( - Any, - { - **({"model": self.__model} if self.__model is not omit else {}), - **extra_query, - }, - ) + params=self.__client._websocket_query_params( + url, + { + **({"model": self.__model} if self.__model is not omit else {}), + **extra_query, + }, ), ) url = url.copy_with(scheme={"http": "ws", "https": "wss"}.get(url.scheme, url.scheme)) @@ -1096,16 +1092,12 @@ def _connect_ws(self, extra_query: Query, extra_headers: Headers) -> WebSocketCo else: url = self._prepare_url() url = url.copy_with( - params=httpx2.QueryParams(self.__client.qs.stringify(cast(Any, self.__client.default_query))) - .merge(url.params) - .merge( - cast( - Any, - { - **({"model": self.__model} if self.__model is not omit else {}), - **extra_query, - }, - ) + params=self.__client._websocket_query_params( + url, + { + **({"model": self.__model} if self.__model is not omit else {}), + **extra_query, + }, ), ) url = url.copy_with(scheme={"http": "ws", "https": "wss"}.get(url.scheme, url.scheme)) diff --git a/src/openai/resources/realtime/translations/translations.py b/src/openai/resources/realtime/translations/translations.py index 8fb860f720..7d2ce9c9f7 100644 --- a/src/openai/resources/realtime/translations/translations.py +++ b/src/openai/resources/realtime/translations/translations.py @@ -538,11 +538,17 @@ async def _connect_ws(self, extra_query: Query, extra_headers: Headers) -> Async raise OpenAIError("You need to install `openai[realtime]` to use this method") from exc url = self._prepare_url() + from ....lib.realtime_translation import _PreparedTranslationQuery + url = url.copy_with( - params=httpx2.QueryParams(self.__client.qs.stringify(cast(Any, self.__client.default_query))) - .merge(url.params) - .merge({"model": self.__model}) - .merge(cast(Any, extra_query)), + params=extra_query + if isinstance(extra_query, _PreparedTranslationQuery) + else self.__client._websocket_query_params( + url, + httpx2.QueryParams({"model": self.__model}).merge(extra_query) + if isinstance(extra_query, httpx2.QueryParams) + else {"model": self.__model, **extra_query}, + ) ) url = url.copy_with(scheme={"http": "ws", "https": "wss"}.get(url.scheme, url.scheme)) options = await self.__client._prepare_options( @@ -948,11 +954,17 @@ def _connect_ws(self, extra_query: Query, extra_headers: Headers) -> WebSocketCo raise OpenAIError("You need to install `openai[realtime]` to use this method") from exc url = self._prepare_url() + from ....lib.realtime_translation import _PreparedTranslationQuery + url = url.copy_with( - params=httpx2.QueryParams(self.__client.qs.stringify(cast(Any, self.__client.default_query))) - .merge(url.params) - .merge({"model": self.__model}) - .merge(cast(Any, extra_query)), + params=extra_query + if isinstance(extra_query, _PreparedTranslationQuery) + else self.__client._websocket_query_params( + url, + httpx2.QueryParams({"model": self.__model}).merge(extra_query) + if isinstance(extra_query, httpx2.QueryParams) + else {"model": self.__model, **extra_query}, + ) ) url = url.copy_with(scheme={"http": "ws", "https": "wss"}.get(url.scheme, url.scheme)) options = self.__client._prepare_options( diff --git a/src/openai/resources/responses/responses.py b/src/openai/resources/responses/responses.py index 54300192eb..f3a2d7321b 100644 --- a/src/openai/resources/responses/responses.py +++ b/src/openai/resources/responses/responses.py @@ -4434,9 +4434,10 @@ async def _connect_ws(self, extra_query: Query, extra_headers: Headers) -> Async url = self._prepare_url() url = url.copy_with( - params=httpx2.QueryParams(self.__client.qs.stringify(cast(Any, self.__client.default_query))) - .merge(url.params) - .merge(cast(Any, extra_query)), + params=self.__client._websocket_query_params( + url, + extra_query, + ), ) url = url.copy_with(scheme={"http": "ws", "https": "wss"}.get(url.scheme, url.scheme)) options = await self.__client._prepare_options( @@ -4839,9 +4840,10 @@ def _connect_ws(self, extra_query: Query, extra_headers: Headers) -> WebSocketCo url = self._prepare_url() url = url.copy_with( - params=httpx2.QueryParams(self.__client.qs.stringify(cast(Any, self.__client.default_query))) - .merge(url.params) - .merge(cast(Any, extra_query)), + params=self.__client._websocket_query_params( + url, + extra_query, + ), ) url = url.copy_with(scheme={"http": "ws", "https": "wss"}.get(url.scheme, url.scheme)) options = self.__client._prepare_options( diff --git a/tests/lib/test_websocket_query_serialization.py b/tests/lib/test_websocket_query_serialization.py new file mode 100644 index 0000000000..0df5cab1ab --- /dev/null +++ b/tests/lib/test_websocket_query_serialization.py @@ -0,0 +1,265 @@ +from __future__ import annotations + +import asyncio +from typing import Any, Literal +from threading import Event, Thread +from urllib.parse import parse_qs, urlsplit + +import httpx2 +import pytest +from websockets.exceptions import ConnectionClosed +from websockets.sync.server import ServerConnection, serve + +from openai import OpenAI, AsyncOpenAI, AzureOpenAI, AsyncAzureOpenAI, omit +from openai._types import Query +from openai.lib.realtime_translation import TranslationSession, AsyncTranslationSession +from tests.lib.test_realtime_translation import script_server +from tests.lib.test_websocket_entry_cleanup import PRODUCTS + + +def connect(client: Any, product: str, query: Query) -> Any: + if product == "realtime": + return client.realtime.connect(model="test", extra_query=query) + if product == "transcription": + return client.realtime.connect(extra_query={"intent": "transcription", **query}) + if product == "sideband": + return client.realtime.connect(call_id="call_1", extra_query=query) + if product == "responses": + return client.responses.connect(extra_query=query) + if product == "beta.responses": + return client.beta.responses.connect(extra_query=query) + if product == "live": + return client.live.connect(extra_query=query) + if product == "live-sideband": + return client.live.sideband.connect(session_id="live_1", extra_query=query) + if product == "live-forks": + return client.live.forks.connect(session_id="live_1", extra_query=query) + if product == "translation": + return client.realtime.translations.connect("test", extra_query=query) + assert product == "beta-realtime" + return client.beta.realtime.connect(model="test", extra_query=query) + + +@pytest.mark.parametrize("mode", ["sync", "async"]) +@pytest.mark.parametrize( + ("provider", "product"), + [("openai", product) for product in (*PRODUCTS, "beta-realtime")] + + [("azure", product) for product in ("realtime", "beta-realtime")], +) +@pytest.mark.parametrize( + ("query", "expected"), + [ + pytest.param( + {"route": "connection", "filter": "plain"}, + {"route": ["connection"], "filter": ["plain"]}, + id="primitive", + ), + pytest.param( + {"route": omit, "filter": omit, "repeat": omit}, + {}, + id="omit", + ), + pytest.param( + {"route": None, "filter": {"language": "fr", "empty": None}, "repeat": ["new", None, "last"]}, + {"route": [""], "filter[language]": ["fr"], "repeat": ["new", "", "last"]}, + id="structured", + ), + ], +) +def test_websocket_query_tiers( + mode: Literal["sync", "async"], + provider: Literal["openai", "azure"], + product: str, + query: Query, + expected: dict[str, list[str]], + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("NO_PROXY", "*") + monkeypatch.setenv("no_proxy", "*") + wire = b'{"type":"future.event","value":"received"}' + observations: list[dict[str, list[str]]] = [] + failures: list[Exception] = [] + done = Event() + original = dict(query) + + def accept(socket: ServerConnection) -> None: + try: + assert socket.request is not None + observations.append(parse_qs(urlsplit(socket.request.path).query, keep_blank_values=True)) + socket.send(wire) + try: + socket.recv(timeout=5) + except ConnectionClosed: + # The client has received the event and is closing normally. + pass + except Exception as exc: + failures.append(exc) + finally: + done.set() + + with serve(accept, "127.0.0.1", 0, close_timeout=1) as server: + thread = Thread(target=server.serve_forever) + thread.start() + try: + address = server.socket.getsockname()[1] + common: dict[str, Any] = { + "api_key": "fake", + "websocket_base_url": ( + f"ws://127.0.0.1:{address}/v1?route=url&repeat=first&repeat=second&keep=one&keep=two" + ), + "default_query": { + "route": "default", + "filter": {"old": "discarded"}, + "global": "kept", + "default_list": ["alpha", "beta"], + "default_absent": None, + }, + } + if provider == "azure": + common.update(azure_endpoint="https://example.invalid/", api_version="test-version") + # Azure's required fields win even if omitted by this request. + actual_query: Query = {**query, "api-version": omit, "deployment": "wrong"} + else: + actual_query = query + if mode == "sync": + cls = OpenAI if provider == "openai" else AzureOpenAI + with cls(**common) as client: + with connect(client, product, actual_query) as connection: + assert connection.recv_bytes() == wire + else: + + async def run() -> None: + cls = AsyncOpenAI if provider == "openai" else AsyncAzureOpenAI + async with cls(**common) as client: + async with connect(client, product, actual_query) as connection: + assert await connection.recv_bytes() == wire + + asyncio.run(run()) + assert done.wait(5), "WebSocket peer did not finish after the client closed" + assert not failures, failures + finally: + server.shutdown() + thread.join(5) + assert not thread.is_alive() + assert len(observations) == 1 + actual = observations[0] + endpoint: dict[str, list[str]] = {} + if provider == "azure": + endpoint = {"api-version": ["test-version"], "deployment": ["test"]} + elif product in {"realtime", "beta-realtime", "translation"}: + endpoint = {"model": ["test"]} + elif product == "sideband": + endpoint = {"call_id": ["call_1"]} + elif product == "transcription": + endpoint = {"intent": ["transcription"]} + repeat = {"repeat": ["first", "second"]} if "repeat" not in query else {} + assert actual == { + "global": ["kept"], + "default_list[]": ["alpha", "beta"], + "keep": ["one", "two"], + **repeat, + **endpoint, + **expected, + } + assert dict(query) == original + + +@pytest.mark.parametrize("mode", ["sync", "async"]) +@pytest.mark.parametrize("managed", [True, False]) +def test_translation_native_query_params_preserve_repeats_and_default_model(mode: str, managed: bool) -> None: + observations: list[dict[str, list[str]]] = [] + + def script(socket: ServerConnection) -> None: + assert socket.request is not None + observations.append(parse_qs(urlsplit(socket.request.path).query)) + socket.send('{"type":"session.closed","event_id":"done"}') + + with script_server(script) as url: + query = httpx2.QueryParams("route=x&route=y") + if mode == "sync": + with OpenAI( + api_key="fake", + base_url=url, + default_query={"inherited": "tenant"}, + http_client=httpx2.Client(trust_env=False), + ) as client: + if managed: + with TranslationSession(client.realtime.translations, model="fixture", extra_query=query) as conn: + assert conn.recv().event_id == "done" + else: + with client.realtime.translations.connect("fixture", extra_query=query) as raw: + assert raw.recv().event_id == "done" + else: + + async def run() -> None: + async with AsyncOpenAI( + api_key="fake", + base_url=url, + default_query={"inherited": "tenant"}, + http_client=httpx2.AsyncClient(trust_env=False), + ) as client: + if managed: + async with AsyncTranslationSession( + client.realtime.translations, model="fixture", extra_query=query + ) as conn: + assert (await asyncio.wait_for(conn.recv(), 5)).event_id == "done" + else: + async with client.realtime.translations.connect("fixture", extra_query=query) as raw: + assert (await asyncio.wait_for(raw.recv(), 5)).event_id == "done" + + asyncio.run(run()) + assert observations == [{"route": ["x", "y"], "model": ["fixture"], "inherited": ["tenant"]}] + assert query.multi_items() == [("route", "x"), ("route", "y")] + + +@pytest.mark.parametrize("mode", ["sync", "async"]) +@pytest.mark.parametrize( + ("query", "expected"), + [ + pytest.param( + {"language": "fr"}, + {"filter[old]": ["discardable"], "route": ["x", "y"], "language": ["fr"]}, + id="untouched", + ), + pytest.param({"route": omit, "filter": omit}, {}, id="omit"), + pytest.param( + {"route": None, "filter": {"new": "value", "empty": None}, "list": ("one", None, "two")}, + {"route": [""], "filter[new]": ["value"], "list": ["one", "", "two"]}, + id="structured", + ), + ], +) +def test_managed_translation_composes_logical_query_without_losing_url_values( + mode: str, query: Query, expected: dict[str, list[str]] +) -> None: + observations: list[dict[str, list[str]]] = [] + + def script(socket: ServerConnection) -> None: + assert socket.request is not None + observations.append(parse_qs(urlsplit(socket.request.path).query, keep_blank_values=True)) + socket.send('{"type":"session.closed","event_id":"done"}') + + original = dict(query) + with script_server(script) as url: + options: dict[str, Any] = { + "api_key": "fake", + "base_url": "https://base.example.invalid/v1?from_http=a&from_http=b&priority=http", + "websocket_base_url": url.replace("http://", "ws://") + "?route=x&route=y&priority=gateway", + "default_query": {"filter": {"old": "discardable"}, "priority": "default"}, + } + if mode == "sync": + with OpenAI(**options) as client: + with TranslationSession(client.realtime.translations, model="fixture", extra_query=query) as conn: + assert conn.recv().event_id == "done" + else: + + async def run() -> None: + async with AsyncOpenAI(**options) as client: + async with AsyncTranslationSession( + client.realtime.translations, model="fixture", extra_query=query + ) as conn: + assert (await asyncio.wait_for(conn.recv(), 5)).event_id == "done" + + asyncio.run(run()) + assert observations == [{"from_http": ["a", "b"], "priority": ["gateway"], "model": ["fixture"], **expected}] + assert dict(query) == original