From 54884ed2d52ec1d50578e60d5e3b6b723b9dd417 Mon Sep 17 00:00:00 2001 From: Arthur Landim Costa Date: Tue, 10 Feb 2026 12:51:24 -0300 Subject: [PATCH 1/3] fix: add aclose() to AsyncStream for standard async cleanup `AsyncStream` exposes `close()` but not `aclose()`, which is the standard Python async cleanup method name (used by contextlib, asyncio, and the language spec for async generators). This causes `AttributeError` when callers use the conventional `aclose()` pattern. Two concrete callers in this repo are affected: - `AsyncChatCompletionStream.close()` stores `raw_stream.response` in `self._response` and calls `self._response.aclose()`. When instrumentation libraries (e.g. Langfuse) wrap the raw stream, the `.response` attribute can resolve to the `AsyncStream` itself rather than the underlying `httpx.Response`, hitting the missing method. - Third-party instrumentation (Langfuse `LangfuseResponseGeneratorAsync`) calls `.aclose()` on the response generator which delegates to the wrapped `AsyncStream`. The fix adds `aclose()` as a thin async alias for `close()`, matching the pattern already used by `httpx.Response`, `asyncio.StreamWriter`, and Python async generators. --- src/openai/_streaming.py | 9 +++++++++ tests/test_streaming.py | 26 ++++++++++++++++++++++++++ 2 files changed, 35 insertions(+) diff --git a/src/openai/_streaming.py b/src/openai/_streaming.py index 61a742668a..a7d354e54a 100644 --- a/src/openai/_streaming.py +++ b/src/openai/_streaming.py @@ -223,6 +223,15 @@ async def close(self) -> None: """ await self.response.aclose() + async def aclose(self) -> None: + """Alias for :meth:`close` following the Python async convention. + + Callers such as ``AsyncChatCompletionStream`` and third-party + instrumentation libraries (e.g. Langfuse) use ``aclose()`` as the + standard async cleanup method. + """ + await self.close() + class ServerSentEvent: def __init__( diff --git a/tests/test_streaming.py b/tests/test_streaming.py index 04f8e51abd..5257a32236 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -216,6 +216,32 @@ def body() -> Iterator[bytes]: assert sse.json() == {"content": "известни"} +@pytest.mark.asyncio +async def test_async_stream_aclose(async_client: AsyncOpenAI) -> None: + """AsyncStream should support aclose() as an alias for close(). + + This is the standard Python async cleanup method name (used by contextlib, + asyncio, and the language spec for async generators). Callers such as + ``AsyncChatCompletionStream.close()`` and Langfuse's + ``LangfuseResponseGeneratorAsync`` invoke ``aclose()`` on the underlying + stream, so its absence causes ``AttributeError`` at cleanup time. + """ + + def body() -> Iterator[bytes]: + yield b"data: [DONE]\n\n" + + stream = AsyncStream( + cast_to=object, + client=async_client, + response=httpx.Response(200, content=to_aiter(body())), + ) + + assert hasattr(stream, "aclose"), "AsyncStream must expose aclose()" + + # aclose() should behave identically to close() + await stream.aclose() + + async def to_aiter(iter: Iterator[bytes]) -> AsyncIterator[bytes]: for chunk in iter: yield chunk From 9e552b381a2865c8abac02921d3515071b928717 Mon Sep 17 00:00:00 2001 From: Arthur Landim Costa Date: Tue, 10 Feb 2026 16:57:29 -0300 Subject: [PATCH 2/3] test: verify aclose() delegates to close() Replace the smoke test with a mock-based assertion that aclose() actually calls close(), validating the behavioral contract. Co-Authored-By: Claude Opus 4.6 --- tests/test_streaming.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/tests/test_streaming.py b/tests/test_streaming.py index 5257a32236..4a566fd5a1 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -1,6 +1,7 @@ from __future__ import annotations from typing import Iterator, AsyncIterator +from unittest import mock import httpx import pytest @@ -238,8 +239,10 @@ def body() -> Iterator[bytes]: assert hasattr(stream, "aclose"), "AsyncStream must expose aclose()" - # aclose() should behave identically to close() - await stream.aclose() + # aclose() should delegate to close() + with mock.patch.object(stream, "close", wraps=stream.close) as mock_close: + await stream.aclose() + mock_close.assert_called_once() async def to_aiter(iter: Iterator[bytes]) -> AsyncIterator[bytes]: From d6c3773c49ef9d2696ce4ab2e6914c9949ae48b8 Mon Sep 17 00:00:00 2001 From: Marcus Wood Date: Thu, 10 Sep 2026 16:21:51 +0000 Subject: [PATCH 3/3] test: verify async stream alias closes responses --- src/openai/_streaming.py | 7 +---- tests/test_streaming.py | 58 ++++++++++++++++++++++++++-------------- 2 files changed, 39 insertions(+), 26 deletions(-) diff --git a/src/openai/_streaming.py b/src/openai/_streaming.py index 34d71d5012..0fd375604d 100644 --- a/src/openai/_streaming.py +++ b/src/openai/_streaming.py @@ -240,12 +240,7 @@ async def close(self) -> None: await self.response.aclose() async def aclose(self) -> None: - """Alias for :meth:`close` following the Python async convention. - - Callers such as ``AsyncChatCompletionStream`` and third-party - instrumentation libraries (e.g. Langfuse) use ``aclose()`` as the - standard async cleanup method. - """ + """Close the response and release the connection. Alias for `close()`.""" await self.close() diff --git a/tests/test_streaming.py b/tests/test_streaming.py index df11c477cb..ee3af47ab6 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -1,7 +1,7 @@ from __future__ import annotations from typing import Iterator, AsyncIterator -from unittest import mock +from contextlib import aclosing, nullcontext import httpx2 import pytest @@ -219,30 +219,48 @@ def body() -> Iterator[bytes]: @pytest.mark.asyncio async def test_async_stream_aclose(async_client: AsyncOpenAI) -> None: - """AsyncStream should support aclose() as an alias for close(). - - This is the standard Python async cleanup method name (used by contextlib, - asyncio, and the language spec for async generators). Callers such as - ``AsyncChatCompletionStream.close()`` and Langfuse's - ``LangfuseResponseGeneratorAsync`` invoke ``aclose()`` on the underlying - stream, so its absence causes ``AttributeError`` at cleanup time. - """ - def body() -> Iterator[bytes]: yield b"data: [DONE]\n\n" - stream = AsyncStream( - cast_to=object, - client=async_client, - response=httpx.Response(200, content=to_aiter(body())), - ) + response = httpx2.Response(200, content=to_aiter(body())) + stream = AsyncStream(cast_to=object, client=async_client, response=response) + + assert not response.is_closed + await stream.aclose() + assert response.is_closed + + # Either spelling remains safe after the response has already been closed. + await stream.close() + await stream.aclose() - assert hasattr(stream, "aclose"), "AsyncStream must expose aclose()" - # aclose() should delegate to close() - with mock.patch.object(stream, "close", wraps=stream.close) as mock_close: - await stream.aclose() - mock_close.assert_called_once() +@pytest.mark.asyncio +@pytest.mark.parametrize("raise_error", [False, True], ids=["early-exit", "exception"]) +async def test_async_stream_aclosing(raise_error: bool) -> None: + def body() -> Iterator[bytes]: + yield ( + b'data: {"id":"chatcmpl-test","object":"chat.completion.chunk","created":0,' + b'"model":"test-model","choices":[{"index":0,"delta":{"content":"hello"},' + b'"finish_reason":null}]}\n\n' + ) + yield b"data: [DONE]\n\n" + + response = httpx2.Response(200, content=to_aiter(body()), headers={"content-type": "text/event-stream"}) + async with AsyncOpenAI( + api_key="fake-test-key", + http_client=httpx2.AsyncClient(transport=httpx2.MockTransport(lambda _request: response)), + ) as client: + stream = await client.chat.completions.create(model="test-model", messages=[], stream=True) + with pytest.raises(ValueError, match="test exception") if raise_error else nullcontext(): + async with aclosing(stream): + async for chunk in stream: + assert chunk.choices[0].delta.content == "hello" + assert not response.is_closed + if raise_error: + raise ValueError("test exception") + break + + assert response.is_closed async def to_aiter(iter: Iterator[bytes]) -> AsyncIterator[bytes]: