Support streaming in ASGITransport

[tjol]

Summary

This changes ASGITransport to support streaming responses, so you can use ASGITransport to test things like SSE.

I wrote this without realizing that PR #1032 by @Kludex, which aims to achieve the same thing, exists. The basic idea is the same: run the app in a separate task.

The reason I'm opening this PR in competition with #1032 is that currently, #1032 breaks some websocket tests (after merging main)

Checklist

• I understand that this PR may be closed in case there was no previous discussion. (This doesn't apply to typos!)
• I've added a test for each change that was introduced, and I tried as much as possible to make a single atomic change.
• I've updated the documentation accordingly.

→ IMHO this doesn't need to be mentioned explicitly anywhere as it just implements the intuitive and expected behaviour for an ASGITransport. It doesn't contradict anything in the current docs that I can see. I could of course add a note to the docs like #1032 if this is deemed necessary.

[codspeed-hq]

Merging this PR will not alter performance

✅ 17 untouched benchmarks
⏩ 7 skipped benchmarks¹

---

_{Comparing tjol:fix-asgi-streaming (20bb5d4) with main (9fdafff)}

Footnotes

1. 7 benchmarks were skipped, so the baseline results were used instead. If they were deleted from the codebase, click here and archive them to remove them from the performance reports. ↩

[cubic-dev-ai]

3 issues found and verified against the latest diff

Prompt for AI agents (unresolved issues)

Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.


<file name="src/httpx2/httpx2/_transports/asgi.py">

<violation number="1" location="src/httpx2/httpx2/_transports/asgi.py:167">
P1: When an app sends the final `http.response.body` with `more_body=False`, the body memory stream is only closed by `app_wrapper`'s `finally`. `ASGIResponseStream.__aiter__` keeps iterating the memory stream and never notices `response_complete`. If the ASGI app does any post-response work (including calling `receive()` again) before returning, `response.aread()`/streaming consumers will block indefinitely even though the response body is complete. Response completion should close the send side of the stream once `more_body` is `False`, not only when the app coroutine finishes.</violation>

<violation number="2" location="src/httpx2/httpx2/_transports/asgi.py:182">
P2: When a streaming response closes before the app exits, this second send raises because the response channel is already closed, causing the shared task group to fail. Suppress broken or closed channel errors while publishing the app exception after consumer disconnect.</violation>

<violation number="3" location="src/httpx2/httpx2/_transports/asgi.py:189">
P1: When callers use `ASGITransport` without an `async with` block, this guard rejects every request, including documented manual `AsyncClient` usage. Make the task-group lifecycle work without `__aenter__`, and close it from `aclose()`.</violation>
</file>

<sub>Reply with feedback, questions, or to request a fix.

Re-trigger cubic</sub>

[cubic-dev-ai]

P1: When callers use ASGITransport without an async with block, this guard rejects every request, including documented manual AsyncClient usage. Make the task-group lifecycle work without __aenter__, and close it from aclose().

Prompt for AI agents

Check if this issue is valid — if so, understand the root cause and fix it. At src/httpx2/httpx2/_transports/asgi.py, line 188:

<comment>When callers use `ASGITransport` without an `async with` block, this guard rejects every request, including documented manual `AsyncClient` usage. Make the task-group lifecycle work without `__aenter__`, and close it from `aclose()`.</comment>

<file context>
@@ -137,42 +148,55 @@ async def receive() -> dict[str, typing.Any]:
+            finally:
+                body_send.close()
+
+        if self._task_group is None:
+            raise RuntimeError("ASGITransport.__aenter__ not called")
+        self._task_group.start_soon(app_wrapper)
</file context>

[tjol]

AsyncClient takes care of this (nowhere in the tests use async with ASGITransport...)

[cubic-dev-ai]

P1: When an app sends the final http.response.body with more_body=False, the body memory stream is only closed by app_wrapper's finally. ASGIResponseStream.__aiter__ keeps iterating the memory stream and never notices response_complete. If the ASGI app does any post-response work (including calling receive() again) before returning, response.aread()/streaming consumers will block indefinitely even though the response body is complete. Response completion should close the send side of the stream once more_body is False, not only when the app coroutine finishes.

Prompt for AI agents

Check if this issue is valid — if so, understand the root cause and fix it. At src/httpx2/httpx2/_transports/asgi.py, line 166:

<comment>When an app sends the final `http.response.body` with `more_body=False`, the body memory stream is only closed by `app_wrapper`'s `finally`. `ASGIResponseStream.__aiter__` keeps iterating the memory stream and never notices `response_complete`. If the ASGI app does any post-response work (including calling `receive()` again) before returning, `response.aread()`/streaming consumers will block indefinitely even though the response body is complete. Response completion should close the send side of the stream once `more_body` is `False`, not only when the app coroutine finishes.</comment>

<file context>
@@ -137,42 +148,55 @@ async def receive() -> dict[str, typing.Any]:
 
                 if body and request.method != "HEAD":
-                    body_parts.append(body)
+                    await body_send.send(body)
 
                 if not more_body:
</file context>

[cubic-dev-ai]

P2: When a streaming response closes before the app exits, this second send raises because the response channel is already closed, causing the shared task group to fail. Suppress broken or closed channel errors while publishing the app exception after consumer disconnect.

Prompt for AI agents

Check if this issue is valid — if so, understand the root cause and fix it. At src/httpx2/httpx2/_transports/asgi.py, line 181:

<comment>When a streaming response closes before the app exits, this second send raises because the response channel is already closed, causing the shared task group to fail. Suppress broken or closed channel errors while publishing the app exception after consumer disconnect.</comment>

<file context>
@@ -137,42 +148,55 @@ async def receive() -> dict[str, typing.Any]:
+                if response_headers is None:
+                    response_headers = {}
+                if self.raise_app_exceptions:
+                    await body_send.send(exc)
+                    app_exception = exc
+                response_started.set()
</file context>

Suggested change

	await body_send.send(exc)
	with contextlib.suppress(anyio.BrokenResourceError, anyio.ClosedResourceError):
	await body_send.send(exc)