From 5544c18c19f125718ed9c4f7c42866c27060158e Mon Sep 17 00:00:00 2001 From: Marcus Wood Date: Wed, 30 Sep 2026 00:06:03 +0000 Subject: [PATCH 1/5] fix: transform NotRequired typed dictionary fields (#3995) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fields wrapped in `NotRequired` currently skip alias, format, and nested transformation metadata. Unwrap it alongside `Required` so optional typed-dictionary fields serialize consistently, while omitted fields stay omitted. Carries forward @Shubham-Padkonde’s fix from #3872 with current main merged in, preserving the original commit and authorship. Includes one focused sync/async regression covering aliases, date formatting, nested dictionaries, lists, and omission. Validation: both regression cases fail on main; 69 transform/typing tests pass on each Pydantic version. Ruff, mypy, strict Pyright, and the custom-code budget pass. Security review found no new dependencies, execution paths, network destinations, or credential handling. Co-authored-by: Shubham Padkonde --- src/openai/_utils/_transform.py | 5 +++-- src/openai/_utils/_typing.py | 9 +++++++-- tests/test_transform.py | 23 ++++++++++++++++++++++- 3 files changed, 32 insertions(+), 5 deletions(-) diff --git a/src/openai/_utils/_transform.py b/src/openai/_utils/_transform.py index 9f1e66ebb5..7316075d4d 100644 --- a/src/openai/_utils/_transform.py +++ b/src/openai/_utils/_transform.py @@ -27,6 +27,7 @@ is_required_type, is_sequence_type, is_annotated_type, + is_not_required_type, strip_annotated_type, ) @@ -117,8 +118,8 @@ def _get_annotated_type(type_: type) -> type | None: This also unwraps the type when applicable, e.g. `Required[Annotated[T, ...]]` """ - if is_required_type(type_): - # Unwrap `Required[Annotated[T, ...]]` to `Annotated[T, ...]` + if is_required_type(type_) or is_not_required_type(type_): + # Unwrap Required or NotRequired to expose the Annotated metadata type_ = get_args(type_)[0] if is_annotated_type(type_): diff --git a/src/openai/_utils/_typing.py b/src/openai/_utils/_typing.py index 193109f3ad..ed43d9d64c 100644 --- a/src/openai/_utils/_typing.py +++ b/src/openai/_utils/_typing.py @@ -9,6 +9,7 @@ TypeIs, Required, Annotated, + NotRequired, get_args, get_origin, ) @@ -45,6 +46,10 @@ def is_required_type(typ: type) -> bool: return get_origin(typ) == Required +def is_not_required_type(typ: type) -> bool: + return get_origin(typ) == NotRequired + + def is_typevar(typ: type) -> bool: # type ignore is required because type checkers # think this expression will always return False @@ -71,10 +76,10 @@ def is_type_alias_type(tp: Any, /) -> TypeIs[typing_extensions.TypeAliasType]: return isinstance(tp, _TYPE_ALIAS_TYPES) -# Extracts T from Annotated[T, ...] or from Required[Annotated[T, ...]] +# Extracts T from Annotated[T, ...], including Required and NotRequired wrappers @lru_cache(maxsize=8096) def strip_annotated_type(typ: type) -> type: - if is_required_type(typ) or is_annotated_type(typ): + if is_required_type(typ) or is_not_required_type(typ) or is_annotated_type(typ): return strip_annotated_type(cast(type, get_args(typ)[0])) return typ diff --git a/tests/test_transform.py b/tests/test_transform.py index 69a5549d9b..8fba1197f6 100644 --- a/tests/test_transform.py +++ b/tests/test_transform.py @@ -4,7 +4,7 @@ import pathlib from typing import Any, Dict, List, Union, TypeVar, Iterable, Optional, cast from datetime import date, datetime -from typing_extensions import Required, Annotated, TypedDict +from typing_extensions import Required, Annotated, TypedDict, NotRequired import pytest @@ -505,3 +505,24 @@ async def test_strips_notgiven(use_async: bool) -> None: async def test_strips_omit(use_async: bool) -> None: assert await transform({"foo_bar": "bar"}, Foo1, use_async) == {"fooBar": "bar"} assert await transform({"foo_bar": omit}, Foo1, use_async) == {} + + +class DateDictWithNotRequiredAlias(TypedDict): + optional_prop: NotRequired[Annotated[date, PropertyInfo(format="iso8601", alias="prop")]] + nested: NotRequired[Bar2] + items: NotRequired[List[Bar2]] + + +@parametrize +@pytest.mark.asyncio +async def test_not_required_transforms(use_async: bool) -> None: + assert await transform(cast(Dict[str, Any], {}), DateDictWithNotRequiredAlias, use_async) == {} + assert await transform({"optional_prop": date(2023, 2, 23)}, DateDictWithNotRequiredAlias, use_async) == { + "prop": "2023-02-23" + } + assert await transform({"nested": {"this_thing": 1}}, DateDictWithNotRequiredAlias, use_async) == { + "nested": {"this__thing": 1} + } + assert await transform({"items": [{"this_thing": 1}]}, DateDictWithNotRequiredAlias, use_async) == { + "items": [{"this__thing": 1}] + } From 841d9297aed9230c29645a446b489ac4e8ccfc5a Mon Sep 17 00:00:00 2001 From: "openai-sdks[bot]" <284451331+openai-sdks[bot]@users.noreply.github.com> Date: Wed, 30 Sep 2026 00:19:39 +0000 Subject: [PATCH 2/5] release: 3.22.1 (#3994) Automated Release PR --- ## [3.22.1](https://github.com/openai/openai-python/compare/v3.22.0...v3.22.1) (2026-09-30) ### Bug Fixes * **api:** correct the missing authentication error message ([#3993](https://github.com/openai/openai-python/issues/3993)) ([51692ed](https://github.com/openai/openai-python/commit/51692ed3f9d2b87084801295f824c2202dac3785)), closes [#3962](https://github.com/openai/openai-python/issues/3962) * transform NotRequired typed dictionary fields ([#3995](https://github.com/openai/openai-python/issues/3995)) ([5544c18](https://github.com/openai/openai-python/commit/5544c18c19f125718ed9c4f7c42866c27060158e)) ### Chores * **tests:** mark sample_file.txt as text with LF line endings ([#3879](https://github.com/openai/openai-python/issues/3879)) ([fd78a48](https://github.com/openai/openai-python/commit/fd78a48d2be489fde978696e1ad32032c20f0fab)) ### Documentation * **examples:** fix two stale example comments ([#3877](https://github.com/openai/openai-python/issues/3877)) ([1c4830b](https://github.com/openai/openai-python/commit/1c4830b6e735d2ca3c4bd95f66467fa007f8b451)) * **examples:** print event.delta in the streaming helpers doc ([#3876](https://github.com/openai/openai-python/issues/3876)) ([063375b](https://github.com/openai/openai-python/commit/063375b7a12f3f3a0de10aec829c0416651be8cc)) --- This PR was generated with [Release Please](https://github.com/googleapis/release-please). See [documentation](https://github.com/googleapis/release-please#release-please). Co-authored-by: openai-sdks[bot] <284451331+openai-sdks[bot]@users.noreply.github.com> --- .release-please-manifest.json | 2 +- CHANGELOG.md | 19 +++++++++++++++++++ pyproject.toml | 2 +- src/openai/_version.py | 2 +- uv.lock | 2 +- 5 files changed, 23 insertions(+), 4 deletions(-) diff --git a/.release-please-manifest.json b/.release-please-manifest.json index 1887bd54c2..918958ff49 100644 --- a/.release-please-manifest.json +++ b/.release-please-manifest.json @@ -1,3 +1,3 @@ { - ".": "3.22.0" + ".": "3.22.1" } \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md index 0914b6b84d..d7b20f5e1d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,24 @@ # Changelog +## [3.22.1](https://github.com/openai/openai-python/compare/v3.22.0...v3.22.1) (2026-09-30) + + +### Bug Fixes + +* **api:** correct the missing authentication error message ([#3993](https://github.com/openai/openai-python/issues/3993)) ([51692ed](https://github.com/openai/openai-python/commit/51692ed3f9d2b87084801295f824c2202dac3785)), closes [#3962](https://github.com/openai/openai-python/issues/3962) +* transform NotRequired typed dictionary fields ([#3995](https://github.com/openai/openai-python/issues/3995)) ([5544c18](https://github.com/openai/openai-python/commit/5544c18c19f125718ed9c4f7c42866c27060158e)) + + +### Chores + +* **tests:** mark sample_file.txt as text with LF line endings ([#3879](https://github.com/openai/openai-python/issues/3879)) ([fd78a48](https://github.com/openai/openai-python/commit/fd78a48d2be489fde978696e1ad32032c20f0fab)) + + +### Documentation + +* **examples:** fix two stale example comments ([#3877](https://github.com/openai/openai-python/issues/3877)) ([1c4830b](https://github.com/openai/openai-python/commit/1c4830b6e735d2ca3c4bd95f66467fa007f8b451)) +* **examples:** print event.delta in the streaming helpers doc ([#3876](https://github.com/openai/openai-python/issues/3876)) ([063375b](https://github.com/openai/openai-python/commit/063375b7a12f3f3a0de10aec829c0416651be8cc)) + ## [3.22.0](https://github.com/openai/openai-python/compare/v3.21.0...v3.22.0) (2026-09-29) diff --git a/pyproject.toml b/pyproject.toml index 6d58e4aae1..405102c63f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "openai" -version = "3.22.0" +version = "3.22.1" description = "The official Python library for the openai API" dynamic = ["readme"] license = "Apache-2.0" diff --git a/src/openai/_version.py b/src/openai/_version.py index 021338e782..6da6bd0785 100644 --- a/src/openai/_version.py +++ b/src/openai/_version.py @@ -1,2 +1,2 @@ __title__ = "openai" -__version__ = "3.22.0" # x-release-please-version +__version__ = "3.22.1" # x-release-please-version diff --git a/uv.lock b/uv.lock index 92e0325807..222d23c225 100644 --- a/uv.lock +++ b/uv.lock @@ -1535,7 +1535,7 @@ wheels = [ [[package]] name = "openai" -version = "3.22.0" # x-release-please-version +version = "3.22.1" # x-release-please-version source = { editable = "." } dependencies = [ { name = "anyio" }, From 58aca1dcfd8d04a3c6352fa2c34b3035ea850f57 Mon Sep 17 00:00:00 2001 From: HughChaw <146055770+Hughhhhcoder@users.noreply.github.com> Date: Wed, 30 Sep 2026 00:51:54 +0000 Subject: [PATCH 3/5] fix: use monotonic clock for file processing timeout (#3748) ## Summary - Use `time.monotonic()` for elapsed-time deadlines in both synchronous and asynchronous file-processing polling. - Add regression coverage for a wall-clock rollback in both resource variants. ## Bug `Files.wait_for_processing()` and its asynchronous counterpart used `time.time()` to measure elapsed time. If the system wall clock moves backwards while a file is still processing, the calculated elapsed time also moves backwards and the timeout can be postponed indefinitely. ### Reproduction With `max_wait_seconds=1`, `poll_interval=5`, and a wall clock that changes from `100` to `90` after the first poll: - Expected: raise the timeout error once the monotonic deadline has elapsed. - Actual before this change: continue polling and enter another five-second sleep. ## Root cause Wall-clock time is adjustable and is not suitable for measuring elapsed durations. ## Fix Use Python's monotonic clock for both the initial timestamp and every deadline check. This does not change the public API, poll interval, terminal states, or error message. ## Validation - `tests/lib/test_file_processing.py`: 15 passed - Ruff check: passed - Ruff format check: passed - Isolated synchronous and asynchronous rollback reproducer: passed Commands used: ```text PYTHONPATH=/tmp/openai-python-file-timeout-deps-20260828:src python -m pytest -p no:cacheprovider -o addopts= --confcutdir=tests/lib --asyncio-mode=auto -q tests/lib/test_file_processing.py python -m ruff check --no-cache src/openai/lib/_files.py tests/lib/test_file_processing.py python -m ruff format --no-cache --check src/openai/lib/_files.py tests/lib/test_file_processing.py ``` --------- Co-authored-by: Marcus Wood --- src/openai/lib/_files.py | 8 ++++---- tests/lib/test_file_processing.py | 16 ++++++++++++---- 2 files changed, 16 insertions(+), 8 deletions(-) diff --git a/src/openai/lib/_files.py b/src/openai/lib/_files.py index ce88c689f2..5b131fdf0c 100644 --- a/src/openai/lib/_files.py +++ b/src/openai/lib/_files.py @@ -19,13 +19,13 @@ def wait_for_file_processing( """Poll a file using the caller's resource and sleep hooks.""" TERMINAL_STATES = {"processed", "error", "deleted"} - start = time.time() + start = time.monotonic() file = files.retrieve(id) while file.status not in TERMINAL_STATES: files._sleep(poll_interval) file = files.retrieve(id) - if time.time() - start > max_wait_seconds: + if time.monotonic() - start > max_wait_seconds: raise RuntimeError( f"Giving up on waiting for file {id} to finish processing after {max_wait_seconds} seconds." ) @@ -43,13 +43,13 @@ async def async_wait_for_file_processing( """Poll a file using the caller's async resource and sleep hooks.""" TERMINAL_STATES = {"processed", "error", "deleted"} - start = time.time() + start = time.monotonic() file = await files.retrieve(id) while file.status not in TERMINAL_STATES: await files._sleep(poll_interval) file = await files.retrieve(id) - if time.time() - start > max_wait_seconds: + if time.monotonic() - start > max_wait_seconds: raise RuntimeError( f"Giving up on waiting for file {id} to finish processing after {max_wait_seconds} seconds." ) diff --git a/tests/lib/test_file_processing.py b/tests/lib/test_file_processing.py index 8ac02acfcc..17445d9633 100644 --- a/tests/lib/test_file_processing.py +++ b/tests/lib/test_file_processing.py @@ -56,15 +56,23 @@ async def test_poll_until_processed(files_resource: Files | AsyncFiles) -> None: sleep.assert_called_once_with(0.25) -async def test_timeout(files_resource: Files | AsyncFiles) -> None: +async def test_timeout_after_wall_clock_rollback(files_resource: Files | AsyncFiles) -> None: with ( mock.patch.object(file_helpers, "time") as clock, mock.patch.object(files_resource, "retrieve", return_value=make_file("uploaded")), - mock.patch.object(files_resource, "_sleep"), + mock.patch.object( + files_resource, + "_sleep", + side_effect=[None, AssertionError("poll continued after the deadline")], + ) as sleep, ): - clock.time.side_effect = [0.0, 11.0] + clock.time.side_effect = [100.0, 90.0] + clock.monotonic.side_effect = [100.0, 102.0] + with pytest.raises(RuntimeError, match=f"Giving up on waiting for file {FILE_ID}"): - await wait(files_resource, max_wait_seconds=10) + await wait(files_resource, poll_interval=5.0, max_wait_seconds=1.0) + + sleep.assert_called_once_with(5.0) async def test_retrieve_error_propagates(files_resource: Files | AsyncFiles) -> None: From 50f95ac0f3563a32c9c2e0415e4701e283f17096 Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Wed, 30 Sep 2026 06:36:30 +0000 Subject: [PATCH 4/5] fix(responses): respect send queue limits during reconnect (#4000) ## Summary Responses WebSocket connections now honor `max_queue_size` during reconnect even when no messages were queued before connecting. ## Changes - Keep the caller's configured send queue for synchronous and asynchronous Responses connections, including beta Responses. - Continue raising `WebSocketQueueFullError` at the configured byte limit and flush admitted messages after reconnect. Co-authored-by: markstuart-oai <323302876+markstuart-oai@users.noreply.github.com> --- .../resources/beta/responses/responses.py | 4 +- src/openai/resources/responses/responses.py | 4 +- tests/test_send_queue_reconnect.py | 85 +++++++++++++++++++ 3 files changed, 89 insertions(+), 4 deletions(-) diff --git a/src/openai/resources/beta/responses/responses.py b/src/openai/resources/beta/responses/responses.py index bfe414eb07..ea52d1701c 100644 --- a/src/openai/resources/beta/responses/responses.py +++ b/src/openai/resources/beta/responses/responses.py @@ -4267,7 +4267,7 @@ def __init__( self._extra_headers = extra_headers self._intentionally_closed = False self._is_reconnecting = False - self._send_queue = send_queue or SendQueue() + self._send_queue = send_queue if send_queue is not None else SendQueue() self._event_handler_registry = EventHandlerRegistry(use_lock=False) self.response = AsyncResponsesResponseResource(self) @@ -4763,7 +4763,7 @@ def __init__( self._extra_headers = extra_headers self._intentionally_closed = False self._is_reconnecting = False - self._send_queue = send_queue or SendQueue() + self._send_queue = send_queue if send_queue is not None else SendQueue() self._event_handler_registry = EventHandlerRegistry(use_lock=True) self.response = ResponsesResponseResource(self) diff --git a/src/openai/resources/responses/responses.py b/src/openai/resources/responses/responses.py index dc428a61ba..e4672df944 100644 --- a/src/openai/resources/responses/responses.py +++ b/src/openai/resources/responses/responses.py @@ -4118,7 +4118,7 @@ def __init__( self._extra_headers = extra_headers self._intentionally_closed = False self._is_reconnecting = False - self._send_queue = send_queue or SendQueue() + self._send_queue = send_queue if send_queue is not None else SendQueue() self._event_handler_registry = EventHandlerRegistry(use_lock=False) self.response = AsyncResponsesResponseResource(self) @@ -4616,7 +4616,7 @@ def __init__( self._close_event = threading.Event() self._connection_lock = threading.Lock() self._is_reconnecting = False - self._send_queue = send_queue or SendQueue() + self._send_queue = send_queue if send_queue is not None else SendQueue() self._event_handler_registry = EventHandlerRegistry(use_lock=True) self.response = ResponsesResponseResource(self) diff --git a/tests/test_send_queue_reconnect.py b/tests/test_send_queue_reconnect.py index 5d48ed5097..49d0fc1430 100644 --- a/tests/test_send_queue_reconnect.py +++ b/tests/test_send_queue_reconnect.py @@ -1,10 +1,14 @@ from __future__ import annotations +import json import asyncio +from typing import Any from unittest.mock import AsyncMock, MagicMock +import httpx2 import pytest +from openai import OpenAI, AsyncOpenAI from openai._exceptions import WebSocketQueueFullError from openai._send_queue import SendQueue from openai.resources.realtime.realtime import RealtimeConnection, AsyncRealtimeConnection @@ -14,6 +18,87 @@ AsyncResponsesConnection as AsyncBetaResponsesConnection, ) +from .lib.test_websocket_redirects import options, resource, reconnect, unexpected_http, async_http_client + + +@pytest.mark.parametrize("name", ["realtime", "responses", "beta.responses"]) +@pytest.mark.parametrize("prequeue", [False, True], ids=["empty", "prequeued"]) +def test_manager_keeps_queue_limit_during_reconnect(monkeypatch: pytest.MonkeyPatch, name: str, prequeue: bool) -> None: + event = {"type": "response.create"} + wire_event = json.dumps(event) + opened, replacement = MagicMock(), MagicMock() + connect = MagicMock(return_value=opened) + monkeypatch.setattr("websockets.sync.client.connect", connect) + with OpenAI( + api_key="fake-key", + websocket_base_url="wss://origin.test", + http_client=httpx2.Client(transport=httpx2.MockTransport(unexpected_http)), + ) as client: + manager = resource(client, name).connect( + **options(client, name), + max_queue_size=len(wire_event.encode("utf-8")), + on_reconnecting=reconnect, + initial_delay=0, + ) + if prequeue: + manager.send(event) + with manager as connection: + if prequeue: + opened.send.assert_called_once_with(wire_event) + else: + opened.send.assert_not_called() + + def reconnect_socket(*_args: Any, **_kwargs: Any) -> MagicMock: + connection.send_raw(wire_event) + with pytest.raises(WebSocketQueueFullError): + connection.send_raw("é") + return replacement + + connect.side_effect = reconnect_socket + assert connection._reconnect(RuntimeError("fake disconnect")) + replacement.send.assert_called_once_with(wire_event) + + +@pytest.mark.parametrize("name", ["realtime", "responses", "beta.responses"]) +@pytest.mark.parametrize("prequeue", [False, True], ids=["empty", "prequeued"]) +@pytest.mark.asyncio +async def test_async_manager_keeps_queue_limit_during_reconnect( + monkeypatch: pytest.MonkeyPatch, name: str, prequeue: bool +) -> None: + event = {"type": "response.create"} + wire_event = json.dumps(event) + opened, replacement = MagicMock(), MagicMock() + opened.send, opened.close = AsyncMock(), AsyncMock() + replacement.send, replacement.close = AsyncMock(), AsyncMock() + connect = AsyncMock(return_value=opened) + monkeypatch.setattr("openai.lib._websocket._WebSocketConnect", connect) + async with AsyncOpenAI( + api_key="fake-key", websocket_base_url="wss://origin.test", http_client=async_http_client() + ) as client: + manager = resource(client, name).connect( + **options(client, name), + max_queue_size=len(wire_event.encode("utf-8")), + on_reconnecting=reconnect, + initial_delay=0, + ) + if prequeue: + manager.send(event) + async with manager as connection: + if prequeue: + opened.send.assert_awaited_once_with(wire_event) + else: + opened.send.assert_not_awaited() + + async def reconnect_socket(*_args: Any, **_kwargs: Any) -> MagicMock: + await connection.send_raw(wire_event) + with pytest.raises(WebSocketQueueFullError): + await connection.send_raw("é") + return replacement + + connect.side_effect = reconnect_socket + assert await connection._reconnect(RuntimeError("fake disconnect")) + replacement.send.assert_awaited_once_with(wire_event) + @pytest.mark.parametrize("connection_type", [RealtimeConnection, ResponsesConnection, BetaResponsesConnection]) def test_reconnect_retries_bounded_send_queue( From 7f203fd5cfd96354524cd07498497b2e27184f95 Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Wed, 30 Sep 2026 06:36:56 +0000 Subject: [PATCH 5/5] fix(api): retain WebSocket endpoint paths and query parameters (#3999) ## Summary Preserve WebSocket endpoint paths and query parameters when using a custom base URL. ## Changes - Keep the Responses endpoint when the base URL includes a query string. - Preserve query parameters on an explicit WebSocket base URL for Responses, Realtime, and Live. - Let connection query parameters override the selected URL, and the selected URL override client defaults. Preserve repeated URL query values. Co-authored-by: markstuart-oai <323302876+markstuart-oai@users.noreply.github.com> --- src/openai/lib/azure.py | 18 ++- .../resources/beta/realtime/realtime.py | 44 ++++--- .../resources/beta/responses/responses.py | 26 +++-- src/openai/resources/live/forks.py | 20 ++-- src/openai/resources/live/live.py | 20 ++-- src/openai/resources/live/sideband.py | 38 +++++-- src/openai/resources/realtime/realtime.py | 38 +++++-- src/openai/resources/responses/responses.py | 26 +++-- tests/lib/test_websocket_urls.py | 107 ++++++++++++++++++ 9 files changed, 251 insertions(+), 86 deletions(-) create mode 100644 tests/lib/test_websocket_urls.py diff --git a/src/openai/lib/azure.py b/src/openai/lib/azure.py index d5f9a01ab0..996d201055 100644 --- a/src/openai/lib/azure.py +++ b/src/openai/lib/azure.py @@ -521,13 +521,18 @@ def _configure_realtime(self, model: str, extra_query: Query) -> tuple[httpx2.UR if self.websocket_base_url is not None: base_url = normalize_httpx_url(self.websocket_base_url) - merge_raw_path = base_url.raw_path.rstrip(b"/") + b"/realtime" + path, separator, url_query = base_url.raw_path.partition(b"?") + merge_raw_path = path.rstrip(b"/") + b"/realtime" + separator + url_query realtime_url = base_url.copy_with(raw_path=merge_raw_path) else: base_url = self._prepare_url("/realtime") realtime_url = base_url.copy_with(scheme="wss") - url = realtime_url.copy_with(params={**query}) + url = realtime_url.copy_with( + params=httpx2.QueryParams(self.qs.stringify(cast(Any, self.default_query))) + .merge(realtime_url.params) + .merge(cast(Any, query)) + ) return url, auth_headers @@ -883,11 +888,16 @@ async def _configure_realtime(self, model: str, extra_query: Query) -> tuple[htt if self.websocket_base_url is not None: base_url = normalize_httpx_url(self.websocket_base_url) - merge_raw_path = base_url.raw_path.rstrip(b"/") + b"/realtime" + path, separator, url_query = base_url.raw_path.partition(b"?") + merge_raw_path = path.rstrip(b"/") + b"/realtime" + separator + url_query realtime_url = base_url.copy_with(raw_path=merge_raw_path) else: base_url = self._prepare_url("/realtime") realtime_url = base_url.copy_with(scheme="wss") - url = realtime_url.copy_with(params={**query}) + url = realtime_url.copy_with( + params=httpx2.QueryParams(self.qs.stringify(cast(Any, self.default_query))) + .merge(realtime_url.params) + .merge(cast(Any, query)) + ) return url, auth_headers diff --git a/src/openai/resources/beta/realtime/realtime.py b/src/openai/resources/beta/realtime/realtime.py index 7124287c0e..99f3d10191 100644 --- a/src/openai/resources/beta/realtime/realtime.py +++ b/src/openai/resources/beta/realtime/realtime.py @@ -364,12 +364,19 @@ async def __aenter__(self) -> AsyncRealtimeConnection: url, auth_headers = await self.__client._configure_realtime(self.__model, extra_query) else: - url = self._prepare_url().copy_with( - params={ - **self.__client.base_url.params, - "model": self.__model, - **extra_query, - }, + 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, + }, + ) + ), ) log.debug("Connecting to WebSocket API") if self.__websocket_connection_options: @@ -400,7 +407,8 @@ def _prepare_url(self) -> httpx2.URL: else: base_url = self.__client._base_url.copy_with(scheme="wss") - merge_raw_path = base_url.raw_path.rstrip(b"/") + b"/realtime" + path, separator, query = base_url.raw_path.partition(b"?") + merge_raw_path = path.rstrip(b"/") + b"/realtime" + separator + query return base_url.copy_with(raw_path=merge_raw_path) async def __aexit__( @@ -547,12 +555,19 @@ def __enter__(self) -> RealtimeConnection: if is_azure_client(self.__client): url, auth_headers = self.__client._configure_realtime(self.__model, extra_query) else: - url = self._prepare_url().copy_with( - params={ - **self.__client.base_url.params, - "model": self.__model, - **extra_query, - }, + 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, + }, + ) + ), ) log.debug("Connecting to WebSocket API") if self.__websocket_connection_options: @@ -583,7 +598,8 @@ def _prepare_url(self) -> httpx2.URL: else: base_url = self.__client._base_url.copy_with(scheme="wss") - merge_raw_path = base_url.raw_path.rstrip(b"/") + b"/realtime" + path, separator, query = base_url.raw_path.partition(b"?") + merge_raw_path = path.rstrip(b"/") + b"/realtime" + separator + query return base_url.copy_with(raw_path=merge_raw_path) def __exit__( diff --git a/src/openai/resources/beta/responses/responses.py b/src/openai/resources/beta/responses/responses.py index ea52d1701c..e42d4ea3b4 100644 --- a/src/openai/resources/beta/responses/responses.py +++ b/src/openai/resources/beta/responses/responses.py @@ -4674,11 +4674,11 @@ async def _connect_ws(self, extra_query: Query, extra_headers: Headers) -> Async except ImportError as exc: raise OpenAIError("You need to install `openai[realtime]` to use this method") from exc - url = self._prepare_url().copy_with( - params={ - **self.__client.base_url.params, - **extra_query, - }, + 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)), ) url = url.copy_with(scheme={"http": "ws", "https": "wss"}.get(url.scheme, url.scheme)) options = await self.__client._prepare_options( @@ -4722,7 +4722,8 @@ def _prepare_url(self) -> httpx2.URL: ws_scheme = "ws" if scheme == "http" else "wss" base_url = self.__client._base_url.copy_with(scheme=ws_scheme) - merge_raw_path = base_url.raw_path.rstrip(b"/") + b"/responses" + path, separator, query = base_url.raw_path.partition(b"?") + merge_raw_path = path.rstrip(b"/") + b"/responses" + separator + query return base_url.copy_with(raw_path=merge_raw_path) async def __aexit__( @@ -5160,11 +5161,11 @@ def _connect_ws(self, extra_query: Query, extra_headers: Headers) -> WebSocketCo except ImportError as exc: raise OpenAIError("You need to install `openai[realtime]` to use this method") from exc - url = self._prepare_url().copy_with( - params={ - **self.__client.base_url.params, - **extra_query, - }, + 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)), ) url = url.copy_with(scheme={"http": "ws", "https": "wss"}.get(url.scheme, url.scheme)) options = self.__client._prepare_options( @@ -5208,7 +5209,8 @@ def _prepare_url(self) -> httpx2.URL: ws_scheme = "ws" if scheme == "http" else "wss" base_url = self.__client._base_url.copy_with(scheme=ws_scheme) - merge_raw_path = base_url.raw_path.rstrip(b"/") + b"/responses" + path, separator, query = base_url.raw_path.partition(b"?") + merge_raw_path = path.rstrip(b"/") + b"/responses" + separator + query return base_url.copy_with(raw_path=merge_raw_path) def __exit__( diff --git a/src/openai/resources/live/forks.py b/src/openai/resources/live/forks.py index 8deb5fbab5..949d4504d7 100644 --- a/src/openai/resources/live/forks.py +++ b/src/openai/resources/live/forks.py @@ -542,11 +542,11 @@ async def _connect_ws(self, extra_query: Query, extra_headers: Headers) -> Async except ImportError as exc: raise OpenAIError("You need to install `openai[realtime]` to use this method") from exc - url = self._prepare_url().copy_with( - params={ - **self.__client.base_url.params, - **extra_query, - }, + 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)), ) url = url.copy_with(scheme={"http": "ws", "https": "wss"}.get(url.scheme, url.scheme)) options = await self.__client._prepare_options( @@ -1025,11 +1025,11 @@ def _connect_ws(self, extra_query: Query, extra_headers: Headers) -> WebSocketCo except ImportError as exc: raise OpenAIError("You need to install `openai[realtime]` to use this method") from exc - url = self._prepare_url().copy_with( - params={ - **self.__client.base_url.params, - **extra_query, - }, + 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)), ) 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 19a78791e1..ec3a54ff52 100644 --- a/src/openai/resources/live/live.py +++ b/src/openai/resources/live/live.py @@ -761,11 +761,11 @@ async def _connect_ws(self, extra_query: Query, extra_headers: Headers) -> Async except ImportError as exc: raise OpenAIError("You need to install `openai[realtime]` to use this method") from exc - url = self._prepare_url().copy_with( - params={ - **self.__client.base_url.params, - **extra_query, - }, + 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)), ) url = url.copy_with(scheme={"http": "ws", "https": "wss"}.get(url.scheme, url.scheme)) options = await self.__client._prepare_options( @@ -1240,11 +1240,11 @@ def _connect_ws(self, extra_query: Query, extra_headers: Headers) -> WebSocketCo except ImportError as exc: raise OpenAIError("You need to install `openai[realtime]` to use this method") from exc - url = self._prepare_url().copy_with( - params={ - **self.__client.base_url.params, - **extra_query, - }, + 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)), ) 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 83256c2c1e..548dae995c 100644 --- a/src/openai/resources/live/sideband.py +++ b/src/openai/resources/live/sideband.py @@ -549,12 +549,19 @@ async def _connect_ws(self, extra_query: Query, extra_headers: Headers) -> Async except ImportError as exc: raise OpenAIError("You need to install `openai[realtime]` to use this method") from exc - url = self._prepare_url().copy_with( - params={ - **self.__client.base_url.params, - **({"graceful_close": self.__graceful_close} if self.__graceful_close is not omit else {}), - **extra_query, - }, + 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, + }, + ) + ), ) url = url.copy_with(scheme={"http": "ws", "https": "wss"}.get(url.scheme, url.scheme)) options = await self.__client._prepare_options( @@ -1037,12 +1044,19 @@ def _connect_ws(self, extra_query: Query, extra_headers: Headers) -> WebSocketCo except ImportError as exc: raise OpenAIError("You need to install `openai[realtime]` to use this method") from exc - url = self._prepare_url().copy_with( - params={ - **self.__client.base_url.params, - **({"graceful_close": self.__graceful_close} if self.__graceful_close is not omit else {}), - **extra_query, - }, + 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, + }, + ) + ), ) 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/realtime/realtime.py b/src/openai/resources/realtime/realtime.py index f541bdd886..6049fd587b 100644 --- a/src/openai/resources/realtime/realtime.py +++ b/src/openai/resources/realtime/realtime.py @@ -717,12 +717,19 @@ async def _connect_ws(self, extra_query: Query, extra_headers: Headers) -> Async url, auth_headers = await self.__client._configure_realtime(model, extra_query) prepared_headers: Headers = extra_headers else: - url = self._prepare_url().copy_with( - params={ - **self.__client.base_url.params, - **({"model": self.__model} if self.__model is not omit else {}), - **extra_query, - }, + 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, + }, + ) + ), ) url = url.copy_with(scheme={"http": "ws", "https": "wss"}.get(url.scheme, url.scheme)) options = await self.__client._prepare_options( @@ -1223,12 +1230,19 @@ def _connect_ws(self, extra_query: Query, extra_headers: Headers) -> WebSocketCo url, auth_headers = self.__client._configure_realtime(model, extra_query) prepared_headers: Headers = extra_headers else: - url = self._prepare_url().copy_with( - params={ - **self.__client.base_url.params, - **({"model": self.__model} if self.__model is not omit else {}), - **extra_query, - }, + 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, + }, + ) + ), ) 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 e4672df944..d7183de8a9 100644 --- a/src/openai/resources/responses/responses.py +++ b/src/openai/resources/responses/responses.py @@ -4525,11 +4525,11 @@ async def _connect_ws(self, extra_query: Query, extra_headers: Headers) -> Async except ImportError as exc: raise OpenAIError("You need to install `openai[realtime]` to use this method") from exc - url = self._prepare_url().copy_with( - params={ - **self.__client.base_url.params, - **extra_query, - }, + 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)), ) url = url.copy_with(scheme={"http": "ws", "https": "wss"}.get(url.scheme, url.scheme)) options = await self.__client._prepare_options( @@ -4573,7 +4573,8 @@ def _prepare_url(self) -> httpx2.URL: ws_scheme = "ws" if scheme == "http" else "wss" base_url = self.__client._base_url.copy_with(scheme=ws_scheme) - merge_raw_path = base_url.raw_path.rstrip(b"/") + b"/responses" + path, separator, query = base_url.raw_path.partition(b"?") + merge_raw_path = path.rstrip(b"/") + b"/responses" + separator + query return base_url.copy_with(raw_path=merge_raw_path) async def __aexit__( @@ -5023,11 +5024,11 @@ def _connect_ws(self, extra_query: Query, extra_headers: Headers) -> WebSocketCo except ImportError as exc: raise OpenAIError("You need to install `openai[realtime]` to use this method") from exc - url = self._prepare_url().copy_with( - params={ - **self.__client.base_url.params, - **extra_query, - }, + 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)), ) url = url.copy_with(scheme={"http": "ws", "https": "wss"}.get(url.scheme, url.scheme)) options = self.__client._prepare_options( @@ -5071,7 +5072,8 @@ def _prepare_url(self) -> httpx2.URL: ws_scheme = "ws" if scheme == "http" else "wss" base_url = self.__client._base_url.copy_with(scheme=ws_scheme) - merge_raw_path = base_url.raw_path.rstrip(b"/") + b"/responses" + path, separator, query = base_url.raw_path.partition(b"?") + merge_raw_path = path.rstrip(b"/") + b"/responses" + separator + query return base_url.copy_with(raw_path=merge_raw_path) def __exit__( diff --git a/tests/lib/test_websocket_urls.py b/tests/lib/test_websocket_urls.py new file mode 100644 index 0000000000..ad360bd4de --- /dev/null +++ b/tests/lib/test_websocket_urls.py @@ -0,0 +1,107 @@ +from __future__ import annotations + +from typing import Any +from urllib.parse import parse_qs, urlsplit + +import httpx2 +import pytest +from websockets.sync.server import ServerConnection + +from openai import OpenAI, AsyncOpenAI, AzureOpenAI, AsyncAzureOpenAI + +from .test_websocket_redirects import resource, reconnect +from .responses.test_websocket_session import script_server + + +@pytest.fixture(autouse=True) +def bypass_proxy(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("NO_PROXY", "127.0.0.1") + monkeypatch.setenv("no_proxy", "127.0.0.1") + + +@pytest.mark.parametrize("mode", ["sync", "async"]) +@pytest.mark.parametrize("with_query", [False, True], ids=["query-free", "with-query"]) +@pytest.mark.parametrize( + "name,source,endpoint", + [ + ("responses", "rest", "/responses"), + ("beta.responses", "rest", "/responses"), + ("responses", "websocket", "/responses"), + ("beta.responses", "websocket", "/responses"), + ("realtime", "websocket", "/realtime"), + ("beta.realtime", "websocket", "/realtime"), + ("live", "websocket", "/live/sessions"), + ("live.sideband", "websocket", "/live/sessions/fake-session/attach"), + ("live.forks", "websocket", "/live/sessions/fake-session/fork"), + ("realtime", "azure", "/realtime"), + ("beta.realtime", "azure", "/realtime"), + ], +) +async def test_actual_upgrade_target(mode: str, with_query: bool, name: str, source: str, endpoint: str) -> None: + received: list[str] = [] + # The legacy beta Realtime API has no reconnect contract. + handshakes = 1 if name == "beta.realtime" else 2 + + def script(socket: ServerConnection) -> None: + assert socket.request is not None + received.append(socket.request.path) + if source == "azure": + assert socket.request.headers.get_all("api-key") == ["fake-websocket-url-test-key"] + socket.send('{"type":"response.future"}') + if len(received) < handshakes: + socket.close(code=1011) + + with script_server(script, expected_connections=handshakes) as url: + host = url.rsplit("/", 1)[0] + query = "?route=x&route=y&priority=url&selected=url" if with_query else "" + client_options: dict[str, Any] = {"api_key": "fake-websocket-url-test-key"} + connect_options: dict[str, Any] = {} + if source == "rest": + client_options["base_url"] = host + "/customer" + query + else: + # None of this REST URL query belongs to the selected WebSocket URL. + client_options["base_url"] = host + "/rest" + ("?rest_only=never" if with_query else "") + client_options["websocket_base_url"] = host.replace("http:", "ws:") + "/customer" + query + + if with_query: + client_options["default_query"] = {"priority": "default", "selected": "default", "client": "kept"} + connect_options["extra_query"] = {"priority": "connection", "request": "kept"} + if source == "azure": + client_options.update(api_version="fake-version") + if with_query: + connect_options["extra_query"].update({"api-version": "ignored", "deployment": "ignored"}) + if "realtime" in name: + connect_options["model"] = "fake-model" + if name.startswith("live."): + connect_options["session_id"] = "fake-session" + if handshakes > 1: + connect_options.update(on_reconnecting=reconnect, initial_delay=0, max_retries=1) + + if mode == "sync": + client_class = AzureOpenAI if source == "azure" else OpenAI + with client_class(**client_options, http_client=httpx2.Client(trust_env=False)) as client: + with resource(client, name).connect(**connect_options) as connection: + events = iter(connection) + for _ in range(handshakes): + assert next(events).type == "response.future" + else: + async_client_class = AsyncAzureOpenAI if source == "azure" else AsyncOpenAI + async with async_client_class( + **client_options, http_client=httpx2.AsyncClient(trust_env=False) + ) as async_client: + async with resource(async_client, name).connect(**connect_options) as async_connection: + async_events = aiter(async_connection) + for _ in range(handshakes): + assert (await anext(async_events)).type == "response.future" + + expected: dict[str, list[str]] = {} + if with_query: + expected.update(route=["x", "y"], priority=["connection"], selected=["url"], client=["kept"], request=["kept"]) + if source == "azure": + expected.update({"api-version": ["fake-version"], "deployment": ["fake-model"]}) + elif "realtime" in name: + expected["model"] = ["fake-model"] + for target in received: + split = urlsplit(target) + assert split.path == "/customer" + endpoint + assert parse_qs(split.query) == expected