StreamingResponse support for AbstractAsyncContextManager[AsyncIterator[bytes]] (TaskGroup support in streaming)

[graingert]

I'd like to be able to return a streaming response driven by a TaskGroup eg:

async def news_and_weather() -> StreamingResponse:
    async def gen():
        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())

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.

I'd like to be able to instead return a new class StreamingCmgrResponse, or pass a cmgr to StreamingResponse:

async def news_and_weather() -> StreamingCmgrResponse:
    @contextlib.asynccontextmanager
    async def acmgr_gen():
        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 StreamingCmgrResponse(acmgr_gen())