Add support for streaming with TaskGroups

[Thomas Grainger]

​https://forum.djangoproject.com/t/streamingresponse-driven-by-a-taskgroup/40320/4
​https://github.com/django/new-features/issues/117

Feature Description

see ​https://forum.djangoproject.com/t/streamingresponse-driven-by-a-taskgroup/40320/4

I’d like to be able to write code that combines multiple streams of data:

async def news_and_weather(request: HttpRequest) -> StreamingHttpResponse:
    async def gen() -> AsyncGenerator[bytes]:
        async def push(ws_url: str, tx: MemoryObjectSendStream) -> None:
            async with tx, connect_ws(ws_url) as conn:
                async for msg in conn:
                    await tx.send(msg)

        async with anyio.create_task_group() as tg:
            tx, rx =  anyio.create_memory_object_stream[bytes]()
            with tx, rx:
                tg.start_soon(push, "ws://example.com/news", tx.clone())
                tg.start_soon(push, "ws://example.com/weather", tx.clone())
                tx.close()
                async for msg in rx:
                    yield msg  # yield in async generator!! illegal inside TaskGroup!
    return StreamingHttpResponse(gen())

Problem

however this doesn’t work because I’m using a yield inside an async generator that’s not a context manager, and calling aclosing() on that async generator is not sufficient to allow a TaskGroup to cancel itself and catch the cancel error.

from useful_types import SupportsAnext

class AsyncIteratorBytesResource(Protocol):
    """
    all the machinery needed to safely run an AsyncGenerator[Bytes]

    (for django-stubs) this allows AsyncGenerator[bytes] but is less strict
    so would also allow a anyio MemoryObjectRecieveStream[bytes]]
    """

    async def __aiter__(self) -> SupportsAnext[bytes]: ...
    async def aclose(self) -> object: ...


async def news_and_weather(request: HttpRequest) -> StreamingAcmgrHttpResponse:
    @contextlib.asynccontextmanager
    async def acmgr_gen() -> AsyncGenerator[AsyncIteratorBytesResource]:
        async def push(ws_url: str, tx: MemoryObjectSendStream) -> None:
            async with tx, connect_ws(ws_url) as conn:
                async for msg in conn:
                    await tx.send(msg)

        async with anyio.create_task_group() as tg:
            tx, rx =  anyio.create_memory_object_stream[bytes]()
            with tx, rx:
                tg.start_soon(push, "ws://example.com/news", tx.clone())
                tg.start_soon(push, "ws://example.com/weather", tx.clone())
                tx.close()
                yield rx  # yield inside asynccontextmanager, permitted inside TaskGroup

    return StreamingAcmgrHttpResponse(acmgr_gen())

Implementation Suggestions

​https://github.com/django/django/pull/19364/changes

[Jacob Walls]

Thanks. Reception on the new-features repo issue seems positive, so though it hasn't moved through any swimlanes there yet, I will speculatively move this one to Accepted assuming it likely will.

[Carlton Gibson]

General approach looks good. I left some comments, and there’s some discussion about a possible alternative approach (but I’m not sure if that’s viable or not, so we should see if that appears.)

[Thomas Grainger]

How about this approach? Working from my phone so posting as a patch rather than pushing directly.

Summary of changes (from Claude)

django/http/response.py

• _set_streaming_content detects async context managers via hasattr(__aenter__, __aexit__), sets self.is_acmgr = True and stores the value in self.__acmgr
• streaming_content property: is_acmgr is an early return at the top (unnested from is_async). Returns an @asynccontextmanager when is_acmgr is True — callers that check is_acmgr use async with response.streaming_content as agen. Both acmgr and regular async paths apply make_bytes and aclose their underlying iterator in a finally block
• StreamingHttpResponse gains __aenter__/__aexit__: for acmgr responses, __aenter__ calls into self.streaming_content (no duplication of awrapper logic needed), stores the context and the yielded generator, then __aexit__ closes the generator and exits the context in the right order so TaskGroup cleans up correctly
• __iter__, __aiter__, and getvalue all raise IsAcmgrException when is_acmgr is True
• IsAcmgrException is exported from django.http

django/core/handlers/asgi.py

• send_response simplified to async with response as content — works for both regular streaming and acmgr responses. aclosing no longer needed

django/middleware/gzip.py

• Adds is_acmgr branch: captures response.streaming_content (an acmgr) and wraps it in a new @asynccontextmanager that feeds the yielded generator into acompress_sequence

django/utils/text.py

• acompress_sequence wraps its entire body in try/finally to aclose its operand on exit if the method is present

django/core/handlers/asgi.py

@@ -4 +4 @@
 import tempfile
 import traceback
 from collections import defaultdict
-from contextlib import aclosing, closing
+from contextlib import closing
 
 from asgiref.sync import ThreadSensitiveContext, sync_to_async
 
@@ -315 +315 @@
         )
         # Streaming responses need to be pinned to their iterator.
         if response.streaming:
-            # - Consume via `__aiter__` and not `streaming_content` directly,
-            #   to allow mapping of a sync iterator.
-            # - Use aclosing() when consuming aiter. See
-            #   https://github.com/python/cpython/commit/6e8dcdaaa49d4313bf9fab9f9923ca5828fbb10e
-            async with aclosing(aiter(response)) as content:
+            async with response as content:
                 async for part in content:
                     for chunk, _ in self.chunk_bytes(part):
                         await send(

django/http/__init__.py

@@ -8 +8 @@
 )
 from django.http.response import (
     BadHeaderError,
+    IsAcmgrException,
     FileResponse,
     Http404,
     HttpResponse,
@@ -47 +48 @@
     "HttpResponseServerError",
     "Http404",
     "BadHeaderError",
+    "IsAcmgrException",
     "JsonResponse",
     "FileResponse",
 ]

django/http/response.py

@@ -7 +7 @@
 import sys
 import time
 import warnings
+from contextlib import asynccontextmanager
 from email.header import Header
 from http.client import responses
 from urllib.parse import urlsplit
@@ -104 +105 @@
     pass
 
 
+class IsAcmgrException(Exception):
+    pass
+
+
 class HttpResponseBase:
     """
     An HTTP response base class with dictionary-accessed headers.
@@ -479 +484 @@
 
     @property
     def streaming_content(self):
+        if self.is_acmgr:
+            # Pull to lexical scope in case streaming_content is set again.
+            _acmgr = self.__acmgr
+
+            @asynccontextmanager
+            async def acmgr_wrapper():
+                async with _acmgr as agen:
+                    try:
+                        async def awrapper():
+                            async for part in agen:
+                                yield self.make_bytes(part)
+
+                        yield awrapper()
+                    finally:
+                        if hasattr(agen, "aclose"):
+                            await agen.aclose()
+
+            return acmgr_wrapper()
         if self.is_async:
             # pull to lexical scope to capture fixed reference in case
             # streaming_content is set again later.
             _iterator = self._iterator
 
             async def awrapper():
-                async for part in _iterator:
-                    yield self.make_bytes(part)
+                try:
+                    async for part in _iterator:
+                        yield self.make_bytes(part)
+                finally:
+                    if hasattr(_iterator, "aclose"):
+                        await _iterator.aclose()
 
             return awrapper()
         else:
@@ -498 +525 @@
 
     def _set_streaming_content(self, value):
         # Ensure we can never iterate on "value" more than once.
+        if hasattr(value, "__aenter__") and hasattr(value, "__aexit__"):
+            self.__acmgr = value
+            self.is_acmgr = True
+            self.is_async = True
+            return
+        self.is_acmgr = False
         try:
             self._iterator = iter(value)
             self.is_async = False
@@ -507 +540 @@
         if hasattr(value, "close"):
             self._resource_closers.append(value.close)
 
+    async def __aenter__(self):
+        if self.is_acmgr:
+            self.__acmgr_ctx = self.streaming_content
+            self.__agen = await self.__acmgr_ctx.__aenter__()
+        else:
+            self.__agen = aiter(self)
+        return self.__agen
+
+    async def __aexit__(self, *exc_info):
+        try:
+            await self.__agen.aclose()
+        finally:
+            if self.is_acmgr:
+                return await self.__acmgr_ctx.__aexit__(*exc_info)
+        return None
+
     def __iter__(self):
+        if self.is_acmgr:
+            raise IsAcmgrException(
+                "%s must be consumed via `async with`. Use `async with response` "
+                "and iterate the yielded content." % self.__class__.__name__
+            )
         try:
             return iter(self.streaming_content)
         except TypeError:
@@ -528 +582 @@
             return map(self.make_bytes, iter(async_to_sync(to_list)(self._iterator)))
 
     async def __aiter__(self):
+        if self.is_acmgr:
+            raise IsAcmgrException(
+                "%s must be consumed via `async with`. Use `async with response` "
+                "and iterate the yielded content." % self.__class__.__name__
+            )
         try:
             async for part in self.streaming_content:
                 yield part
@@ -544 +603 @@
                 yield part
 
     def getvalue(self):
+        if self.is_acmgr:
+            raise IsAcmgrException(
+                "%s must be consumed via `async with`. Use `async with response` "
+                "and iterate the yielded content." % self.__class__.__name__
+            )
         return b"".join(self.streaming_content)
 
 

django/middleware/gzip.py

@@ +1 @@
+from contextlib import asynccontextmanager
+
 from django.utils.cache import patch_vary_headers
 from django.utils.deprecation import MiddlewareMixin
 from django.utils.regex_helper import _lazy_re_compile
@@ -31 +33 @@
             return response
 
         if response.streaming:
-            if response.is_async:
+            if response.is_acmgr:
+                original_acmgr = response.streaming_content
+                max_random_bytes = self.max_random_bytes
+
+                @asynccontextmanager
+                async def compressed_acmgr():
+                    async with original_acmgr as agen:
+                        yield acompress_sequence(
+                            agen,
+                            max_random_bytes=max_random_bytes,
+                        )
+
+                response.streaming_content = compressed_acmgr()
+            elif response.is_async:
                 response.streaming_content = acompress_sequence(
                     response.streaming_content,
                     max_random_bytes=self.max_random_bytes,

django/utils/text.py

@@ -390 +390 @@
 
 
 async def acompress_sequence(sequence, *, max_random_bytes=None):
-    buf = StreamingBuffer()
-    filename = _get_random_filename(max_random_bytes) if max_random_bytes else None
-    with GzipFile(
-        filename=filename, mode="wb", compresslevel=6, fileobj=buf, mtime=0
-    ) as zfile:
-        # Output headers...
+    try:
+        buf = StreamingBuffer()
+        filename = _get_random_filename(max_random_bytes) if max_random_bytes else None
+        with GzipFile(
+            filename=filename, mode="wb", compresslevel=6, fileobj=buf, mtime=0
+        ) as zfile:
+            # Output headers...
+            yield buf.read()
+            async for item in sequence:
+                zfile.write(item)
+                zfile.flush()
+                data = buf.read()
+                if data:
+                    yield data
         yield buf.read()
-        async for item in sequence:
-            zfile.write(item)
-            zfile.flush()
-            data = buf.read()
-            if data:
-                yield data
-    yield buf.read()
+    finally:
+        if hasattr(sequence, "aclose"):
+            await sequence.aclose()

[Thomas Grainger]

Attaching updated patch based on the approach discussed in ​PR #19364.

Summary of changes

django/http/response.py:

• _set_streaming_content detects async context managers via hasattr(__aenter__, __aexit__), sets is_acmgr = True and stores the value
• streaming_content property returns an @asynccontextmanager wrapping make_bytes when is_acmgr is True
• StreamingHttpResponse gains __aenter__/__aexit__: for acmgr responses, enters the streaming_content CM; for regular responses, returns aiter(self). __aexit__ only handles acmgr CM cleanup — aclosing in the ASGI handler handles iterator close
• __iter__, __aiter__, and getvalue raise IsAcmgrException when is_acmgr is True

django/core/handlers/asgi.py:

• Single code path: async with response as agen, aclosing(agen) as content: — works for both regular streaming and acmgr responses

django/middleware/gzip.py:

• Adds is_acmgr branch that wraps the acmgr in a new @asynccontextmanager feeding into acompress_sequence

django/utils/text.py:

• acompress_sequence wraps its body in try/finally to aclose its operand

Tests

• 13 new tests in tests/httpwrappers/tests.py covering: basic acmgr streaming, make_bytes coercion, IsAcmgrException guards on __iter__/__aiter__/getvalue, __aexit__ on error and break, non-acmgr __aenter__ fallback, reassignment, single-producer TaskGroup+Queue, and multi-producer fan-in (news-and-weather pattern)
• 1 new test in tests/middleware/tests.py for gzip compression of acmgr streaming responses

Docs

• New "Streaming with TaskGroup" section in docs/ref/request-response.txt with full news_and_weather example, key points re PEP 789, and anyio note
• Release note in docs/releases/6.1.txt

Patch

django/core/handlers/asgi.py

@@ -318 +318 @@
         )
         # Streaming responses need to be pinned to their iterator.
         if response.streaming:
-            # - Consume via `__aiter__` and not `streaming_content` directly,
-            #   to allow mapping of a sync iterator.
-            # - Use aclosing() when consuming aiter. See
-            #   https://github.com/python/cpython/commit/6e8dcdaaa49d4313bf9fab9f9923ca5828fbb10e
-            async with aclosing(aiter(response)) as content:
+            # Use aclosing() when consuming aiter. See
+            # https://github.com/python/cpython/commit/6e8dcdaaa49d4313bf9fab9f9923ca5828fbb10e
+            async with response as agen, aclosing(agen) as content:
                 async for part in content:
                     for chunk, _ in self.chunk_bytes(part):
                         await send(

django/http/__init__.py

@@ -8 +8 @@
 )
 from django.http.response import (
     BadHeaderError,
+    IsAcmgrException,
     FileResponse,
     Http404,
     HttpResponse,
@@ -47 +48 @@
     "HttpResponseServerError",
     "Http404",
     "BadHeaderError",
+    "IsAcmgrException",
     "JsonResponse",
     "FileResponse",
 ]

django/http/response.py

@@ -7 +7 @@
 import sys
 import time
 import warnings
+from contextlib import asynccontextmanager
 from email.header import Header
 from http.client import responses
 from urllib.parse import urlsplit
@@ -104 +105 @@
     pass
 
 
+class IsAcmgrException(Exception):
+    pass
+
+
 class HttpResponseBase:
     """
     An HTTP response base class with dictionary-accessed headers.
@@ -479 +484 @@
 
     @property
     def streaming_content(self):
+        if self.is_acmgr:
+            # Pull to lexical scope in case streaming_content is set again.
+            _acmgr = self.__acmgr
+
+            @asynccontextmanager
+            async def acmgr_wrapper():
+                async with _acmgr as agen:
+                    async def awrapper():
+                        try:
+                            async for part in agen:
+                                yield self.make_bytes(part)
+                        finally:
+                            if hasattr(agen, "aclose"):
+                                await agen.aclose()
+
+                    yield awrapper()
+
+            return acmgr_wrapper()
         if self.is_async:
             # pull to lexical scope to capture fixed reference in case
             # streaming_content is set again later.
             _iterator = self._iterator
 
             async def awrapper():
-                async for part in _iterator:
-                    yield self.make_bytes(part)
+                try:
+                    async for part in _iterator:
+                        yield self.make_bytes(part)
+                finally:
+                    if hasattr(_iterator, "aclose"):
+                        await _iterator.aclose()
 
             return awrapper()
         else:
@@ -498 +525 @@
 
     def _set_streaming_content(self, value):
         # Ensure we can never iterate on "value" more than once.
+        if hasattr(value, "__aenter__") and hasattr(value, "__aexit__"):
+            self.__acmgr = value
+            self.is_acmgr = True
+            self.is_async = True
+            return
+        self.is_acmgr = False
         try:
             self._iterator = iter(value)
             self.is_async = False
@@ -507 +540 @@
         if hasattr(value, "close"):
             self._resource_closers.append(value.close)
 
+    async def __aenter__(self):
+        if self.is_acmgr:
+            self.__acmgr_ctx = self.streaming_content
+            return await self.__acmgr_ctx.__aenter__()
+        return aiter(self)
+
+    async def __aexit__(self, *exc_info):
+        if self.is_acmgr:
+            return await self.__acmgr_ctx.__aexit__(*exc_info)
+
     def __iter__(self):
+        if self.is_acmgr:
+            raise IsAcmgrException(
+                "%s must be consumed via `async with`. Use `async with response` "
+                "and iterate the yielded content." % self.__class__.__name__
+            )
         try:
             return iter(self.streaming_content)
         except TypeError:
@@ -528 +576 @@
             return map(self.make_bytes, iter(async_to_sync(to_list)(self._iterator)))
 
     async def __aiter__(self):
+        if self.is_acmgr:
+            raise IsAcmgrException(
+                "%s must be consumed via `async with`. Use `async with response` "
+                "and iterate the yielded content." % self.__class__.__name__
+            )
         try:
             async for part in self.streaming_content:
                 yield part
@@ -544 +597 @@
                 yield part
 
     def getvalue(self):
+        if self.is_acmgr:
+            raise IsAcmgrException(
+                "%s must be consumed via `async with`. Use `async with response` "
+                "and iterate the yielded content." % self.__class__.__name__
+            )
         return b"".join(self.streaming_content)
 
 

django/middleware/gzip.py

@@ +1 @@
+from contextlib import asynccontextmanager
+
 from django.utils.cache import patch_vary_headers
 from django.utils.deprecation import MiddlewareMixin
 from django.utils.regex_helper import _lazy_re_compile
@@ -31 +33 @@
             return response
 
         if response.streaming:
-            if response.is_async:
+            if response.is_acmgr:
+                original_acmgr = response.streaming_content
+                max_random_bytes = self.max_random_bytes
+
+                @asynccontextmanager
+                async def compressed_acmgr():
+                    async with original_acmgr as agen:
+                        yield acompress_sequence(
+                            agen,
+                            max_random_bytes=max_random_bytes,
+                        )
+
+                response.streaming_content = compressed_acmgr()
+            elif response.is_async:
                 response.streaming_content = acompress_sequence(
                     response.streaming_content,
                     max_random_bytes=self.max_random_bytes,

django/utils/text.py

@@ -390 +390 @@
 
 
 async def acompress_sequence(sequence, *, max_random_bytes=None):
-    buf = StreamingBuffer()
-    filename = _get_random_filename(max_random_bytes) if max_random_bytes else None
-    with GzipFile(
-        filename=filename, mode="wb", compresslevel=6, fileobj=buf, mtime=0
-    ) as zfile:
-        # Output headers...
+    try:
+        buf = StreamingBuffer()
+        filename = _get_random_filename(max_random_bytes) if max_random_bytes else None
+        with GzipFile(
+            filename=filename, mode="wb", compresslevel=6, fileobj=buf, mtime=0
+        ) as zfile:
+            # Output headers...
+            yield buf.read()
+            async for item in sequence:
+                zfile.write(item)
+                zfile.flush()
+                data = buf.read()
+                if data:
+                    yield data
         yield buf.read()
-        async for item in sequence:
-            zfile.write(item)
-            zfile.flush()
-            data = buf.read()
-            if data:
-                yield data
-    yield buf.read()
+    finally:
+        if hasattr(sequence, "aclose"):
+            await sequence.aclose()
 
 
 # Expression to match some_token and some_token="with spaces" (and similarly

docs/ref/request-response.txt

@@ -1424 +1424 @@
 returning the ``StreamingHttpResponse`` object, then you may also want to
 :ref:`handle disconnections in the view <async-handling-disconnect>` itself.
 
+.. _request-response-streaming-task-groups:
+
+Streaming with ``TaskGroup``
+----------------------------
+
+.. versionadded:: 6.1
+
+When serving under ASGI you may want to stream content that is produced by
+background tasks — for example, fan-in from multiple websocket connections or
+long-running computations. Python's :class:`asyncio.TaskGroup` is the natural
+way to manage those tasks, but an ``async for`` loop cannot ``yield`` inside a
+``TaskGroup`` context (see :pep:`789` for details).
+
+The solution is to pass an :func:`~contextlib.asynccontextmanager` instance as
+the ``streaming_content`` argument. The context manager's ``__aenter__`` sets up
+the ``TaskGroup`` (and any other resources), then ``yield`` s an async iterator
+that the framework will consume. When the response finishes — or the client
+disconnects — the context manager's ``__aexit__`` tears everything down in the
+correct order.
+
+Here is a complete example that merges two websocket feeds into a single
+streaming response::
+
+    import asyncio
+    from collections.abc import AsyncGenerator
+    from contextlib import asynccontextmanager
+
+    from django.http import StreamingHttpResponse
+
+
+    @asynccontextmanager
+    async def news_and_weather() -> AsyncGenerator[AsyncGenerator[bytes, None], None]:
+        queue: asyncio.Queue[bytes | None] = asyncio.Queue(maxsize=1)
+
+        async def consume_ws(url: str) -> None:
+            async with await connect_ws(url) as ws:
+                async for msg in ws:
+                    await queue.put(msg)
+
+        async def run_producers() -> None:
+            async with asyncio.TaskGroup() as tg:
+                tg.create_task(consume_ws("ws://news.example.com/feed"))
+                tg.create_task(consume_ws("ws://weather.example.com/feed"))
+            await queue.put(None)  # sentinel: all producers finished
+
+        async def items() -> AsyncGenerator[bytes, None]:
+            while (item := await queue.get()) is not None:
+                yield item
+
+        async with asyncio.TaskGroup() as tg:
+            tg.create_task(run_producers())
+            yield items()
+
+
+    async def news_and_weather_view(request):
+        return StreamingHttpResponse(news_and_weather())
+
+Key points:
+
+* ``news_and_weather()`` is decorated with
+  :func:`~contextlib.asynccontextmanager`. Django detects this via
+  ``hasattr(value, "__aenter__")`` in ``_set_streaming_content`` and sets
+  :attr:`~StreamingHttpResponse.is_acmgr` to ``True``.
+
+* The ``yield`` that suspends the frame happens inside
+  ``@asynccontextmanager``, which correctly routes exceptions back to the
+  ``TaskGroup`` cancel scope. The ``items()`` async generator yielded *out* to
+  the caller never yields inside a cancel scope, so it is safe.
+
+* Middleware that wraps ``streaming_content`` (such as
+  :class:`~django.middleware.gzip.GZipMiddleware`) is aware of the
+  ``is_acmgr`` flag and preserves the context-manager protocol.
+
+* Attempting to iterate an ``is_acmgr`` response directly (via ``__iter__``,
+  ``__aiter__``, or ``getvalue()``) raises
+  :exc:`~django.http.IsAcmgrException`. The response must be consumed with
+  ``async with``::
+
+      async with response as agen:
+          async for chunk in agen:
+              ...
+
+.. note::
+
+    The `anyio <https://anyio.readthedocs.io/>`_ library provides
+    :func:`anyio.create_task_group` and :func:`anyio.create_memory_object_stream`,
+    which can be used as an alternative to :class:`asyncio.TaskGroup` and
+    :class:`asyncio.Queue`. A ``MemoryObjectReceiveStream`` is an async
+    iterable and can be yielded directly from the context manager.
+
+.. attribute:: StreamingHttpResponse.is_acmgr
+
+    Boolean indicating whether the response was created with an async context
+    manager as its streaming content. When ``True``,
+    :attr:`~StreamingHttpResponse.is_async` is also ``True`` and the response
+    must be consumed via ``async with``.
+
 ``FileResponse`` objects
 ========================
 

docs/releases/6.1.txt

@@ -323 +323 @@
 Requests and Responses
 ~~~~~~~~~~~~~~~~~~~~~~
 
+* :class:`~django.http.StreamingHttpResponse` now accepts an async context
+  manager as ``streaming_content``, enabling streaming patterns that require
+  :class:`asyncio.TaskGroup` or similar structured concurrency primitives. See
+  :ref:`request-response-streaming-task-groups` for details.
+
 * :attr:`HttpRequest.multipart_parser_class <django.http.HttpRequest.multipart_parser_class>`
   can now be customized to use a different multipart parser class.
 

tests/asgi/urls.py

@@ -1 +1 @@
 import asyncio
+import contextlib
 import threading
 import time
 

tests/httpwrappers/tests.py

@@ +1 @@
+import asyncio
+import contextlib
 import copy
 import json
 import os
 import pickle
+import sys
 import unittest
 import uuid
 
@@ -16 +19 @@
     HttpResponseNotModified,
     HttpResponsePermanentRedirect,
     HttpResponseRedirect,
+    IsAcmgrException,
     JsonResponse,
     QueryDict,
     SimpleCookie,
@@ -808 +812 @@
         with self.assertWarnsMessage(Warning, msg):
             self.assertEqual(b"hello", await anext(aiter(r)))
 
+    async def test_async_context_manager_streaming_response(self):
+        """StreamingHttpResponse accepts an async context manager as content."""
+
+        @contextlib.asynccontextmanager
+        async def acmgr():
+            async def iterator():
+                yield b"hello"
+                yield b"world"
+
+            yield iterator()
+
+        r = StreamingHttpResponse(acmgr())
+        self.assertTrue(r.is_acmgr)
+        self.assertTrue(r.is_async)
+
+        chunks = []
+        async with r as agen, contextlib.aclosing(agen) as content:
+            async for chunk in content:
+                chunks.append(chunk)
+        self.assertEqual(chunks, [b"hello", b"world"])
+
+    async def test_acmgr_make_bytes(self):
+        """Acmgr streaming content is coerced to bytes via make_bytes."""
+
+        @contextlib.asynccontextmanager
+        async def acmgr():
+            async def iterator():
+                yield "hello"
+                yield "café"
+
+            yield iterator()
+
+        r = StreamingHttpResponse(acmgr())
+        chunks = []
+        async with r as agen, contextlib.aclosing(agen) as content:
+            async for chunk in content:
+                chunks.append(chunk)
+        self.assertEqual(chunks, [b"hello", b"caf\xc3\xa9"])
+
+    async def test_acmgr_iter_raises(self):
+        """Iterating an acmgr response via __iter__ raises IsAcmgrException."""
+
+        @contextlib.asynccontextmanager
+        async def acmgr():
+            yield iter([b"hello"])
+
+        r = StreamingHttpResponse(acmgr())
+        msg = (
+            "StreamingHttpResponse must be consumed via `async with`. "
+            "Use `async with response` and iterate the yielded content."
+        )
+        with self.assertRaisesMessage(IsAcmgrException, msg):
+            iter(r)
+
+    async def test_acmgr_aiter_raises(self):
+        """Iterating an acmgr response via __aiter__ raises IsAcmgrException."""
+
+        @contextlib.asynccontextmanager
+        async def acmgr():
+            yield iter([b"hello"])
+
+        r = StreamingHttpResponse(acmgr())
+        msg = (
+            "StreamingHttpResponse must be consumed via `async with`. "
+            "Use `async with response` and iterate the yielded content."
+        )
+        with self.assertRaisesMessage(IsAcmgrException, msg):
+            async for _ in r:
+                pass
+
+    async def test_acmgr_getvalue_raises(self):
+        """Calling getvalue() on an acmgr response raises IsAcmgrException."""
+
+        @contextlib.asynccontextmanager
+        async def acmgr():
+            yield iter([b"hello"])
+
+        r = StreamingHttpResponse(acmgr())
+        with self.assertRaises(IsAcmgrException):
+            r.getvalue()
+
+    async def test_acmgr_aexit_called_on_error(self):
+        """The acmgr's __aexit__ is called even when iteration raises."""
+        exited = []
+
+        @contextlib.asynccontextmanager
+        async def acmgr():
+            try:
+                async def iterator():
+                    yield b"hello"
+                    raise ValueError("boom")
+
+                yield iterator()
+            finally:
+                exited.append(True)
+
+        r = StreamingHttpResponse(acmgr())
+        with self.assertRaises(ValueError):
+            async with r as agen, contextlib.aclosing(agen) as content:
+                async for _ in content:
+                    pass
+        self.assertEqual(exited, [True])
+
+    async def test_acmgr_aexit_called_on_break(self):
+        """The acmgr cleans up when the consumer breaks out early."""
+        exited = []
+
+        @contextlib.asynccontextmanager
+        async def acmgr():
+            try:
+                async def iterator():
+                    yield b"hello"
+                    yield b"world"
+
+                yield iterator()
+            finally:
+                exited.append(True)
+
+        r = StreamingHttpResponse(acmgr())
+        async with r as agen, contextlib.aclosing(agen) as content:
+            async for _ in content:
+                break
+        self.assertEqual(exited, [True])
+
+    async def test_acmgr_non_acmgr_aenter_returns_aiter(self):
+        """__aenter__ on a non-acmgr async response returns aiter(self)."""
+
+        async def async_iter():
+            yield b"hello"
+
+        r = StreamingHttpResponse(async_iter())
+        self.assertFalse(r.is_acmgr)
+        agen = await r.__aenter__()
+        chunk = await agen.__anext__()
+        self.assertEqual(chunk, b"hello")
+        await r.__aexit__(None, None, None)
+
+    async def test_acmgr_reassign_streaming_content(self):
+        """Assigning a new acmgr to streaming_content replaces the old one."""
+
+        @contextlib.asynccontextmanager
+        async def acmgr1():
+            async def it():
+                yield b"first"
+
+            yield it()
+
+        @contextlib.asynccontextmanager
+        async def acmgr2():
+            async def it():
+                yield b"second"
+
+            yield it()
+
+        r = StreamingHttpResponse(acmgr1())
+        r.streaming_content = acmgr2()
+        self.assertTrue(r.is_acmgr)
+
+        chunks = []
+        async with r as agen, contextlib.aclosing(agen) as content:
+            async for chunk in content:
+                chunks.append(chunk)
+        self.assertEqual(chunks, [b"second"])
+
+    @unittest.skipUnless(
+        sys.version_info >= (3, 11), "asyncio.TaskGroup requires Python 3.11+"
+    )
+    async def test_taskgroup_queue_streaming_response(self):
+        """
+        StreamingHttpResponse works with asyncio.TaskGroup and asyncio.Queue,
+        the canonical pattern for background-task-driven streaming.
+        """
+
+        @contextlib.asynccontextmanager
+        async def taskgroup_stream():
+            queue = asyncio.Queue()
+
+            async def producer():
+                for chunk in [b"hello", b"world"]:
+                    await queue.put(chunk)
+                await queue.put(None)  # sentinel
+
+            async def generator():
+                while True:
+                    chunk = await queue.get()
+                    if chunk is None:
+                        break
+                    yield chunk
+
+            async with asyncio.TaskGroup() as tg:
+                tg.create_task(producer())
+                yield generator()
+
+        r = StreamingHttpResponse(taskgroup_stream())
+        self.assertTrue(r.is_acmgr)
+
+        chunks = []
+        async with r as agen, contextlib.aclosing(agen) as content:
+            async for chunk in content:
+                chunks.append(chunk)
+        self.assertEqual(chunks, [b"hello", b"world"])
+
+    @unittest.skipUnless(
+        sys.version_info >= (3, 11), "asyncio.TaskGroup requires Python 3.11+"
+    )
+    async def test_taskgroup_multiple_producers_streaming_response(self):
+        """
+        StreamingHttpResponse works with multiple producer tasks feeding a
+        shared queue, as in the news-and-weather websocket fan-in pattern.
+        """
+
+        @contextlib.asynccontextmanager
+        async def multi_producer_stream():
+            queue = asyncio.Queue(maxsize=1)
+
+            async def consume(source):
+                """Simulate a websocket consumer pushing to the shared queue."""
+                for msg in source:
+                    await queue.put(msg)
+
+            async def run_producers():
+                async with asyncio.TaskGroup() as tg:
+                    tg.create_task(consume([b"news-1", b"news-2"]))
+                    tg.create_task(consume([b"weather-1"]))
+                await queue.put(None)  # sentinel after all producers finish
+
+            async def items():
+                while (item := await queue.get()) is not None:
+                    yield item
+
+            async with asyncio.TaskGroup() as tg:
+                tg.create_task(run_producers())
+                yield items()
+
+        r = StreamingHttpResponse(multi_producer_stream())
+        self.assertTrue(r.is_acmgr)
+
+        chunks = []
+        async with r as agen, contextlib.aclosing(agen) as content:
+            async for chunk in content:
+                chunks.append(chunk)
+        # All three messages arrive (order depends on scheduling).
+        self.assertEqual(sorted(chunks), [b"news-1", b"news-2", b"weather-1"])
+
     def test_text_attribute_error(self):
         r = StreamingHttpResponse(iter(["hello", "world"]))
         msg = "This %s instance has no `text` attribute." % r.__class__.__name__

tests/middleware/tests.py

@@ +1 @@
+import contextlib
 import gzip
 import random
 import re
@@ -936 +937 @@
         self.assertEqual(r.get("Content-Encoding"), "gzip")
         self.assertFalse(r.has_header("Content-Length"))
 
+    async def test_compress_acmgr_streaming_response(self):
+        """
+        Compression is performed on responses with async context manager content.
+        """
+
+        async def get_stream_response(request):
+            @contextlib.asynccontextmanager
+            async def acmgr():
+                async def iterator():
+                    for chunk in self.sequence:
+                        yield chunk
+
+                yield iterator()
+
+            resp = StreamingHttpResponse(acmgr())
+            resp["Content-Type"] = "text/html; charset=UTF-8"
+            return resp
+
+        r = await GZipMiddleware(get_stream_response)(self.req)
+        self.assertTrue(r.is_acmgr)
+        chunks = []
+        async with r as agen, contextlib.aclosing(agen) as content:
+            async for chunk in content:
+                chunks.append(chunk)
+        self.assertEqual(
+            self.decompress(b"".join(chunks)),
+            b"".join(self.sequence),
+        )
+        self.assertEqual(r.get("Content-Encoding"), "gzip")
+        self.assertFalse(r.has_header("Content-Length"))
+
     def test_compress_streaming_response_unicode(self):
         """
         Compression is performed on responses with streaming Unicode content.