Opened 104 minutes ago
Last modified 24 minutes ago
#36916 new Uncategorized
add support for streaming with TaskGroups
| Reported by: | Thomas Grainger | Owned by: | |
|---|---|---|---|
| Component: | HTTP handling | Version: | dev |
| Severity: | Normal | Keywords: | structured-concurrency, taskgroups |
| Cc: | Thomas Grainger | Triage Stage: | Unreviewed |
| Has patch: | yes | Needs documentation: | no |
| Needs tests: | no | Patch needs improvement: | no |
| Easy pickings: | yes | UI/UX: | no |
Description (last modified by )
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:
`py
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.
`py
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
Change History (2)
comment:1 by , 38 minutes ago
| Description: | modified (diff) |
|---|---|
| Owner: | set to |
| Status: | new → assigned |
comment:2 by , 24 minutes ago
| Owner: | removed |
|---|---|
| Status: | assigned → new |