Class StreamingResponse

starlette/responses.py:222–283

class StreamingResponse(Response):
    body_iterator: AsyncContentStream

    def __init__(
        self,
        content: ContentStream,
        status_code: int = 200,
        headers: Mapping[str, str] | None = None,
        media_type: str | None = None,
        background: BackgroundTask | None = None,
    ) -> None:
        if isinstance(content, AsyncIterable):
            self.body_iterator = content
        else:
            self.body_iterator = iterate_in_threadpool(content)
        self.status_code = status_code
        self.media_type = self.media_type if media_type is None else media_type
        self.background = background
        self.init_headers(headers)

    async def listen_for_disconnect(self, receive: Receive) -> None:
        while True:
            message = await receive()
            if message[class="st">"type"] == class="st">"http.disconnect":
                break

    async def stream_response(self, send: Send) -> None:
        await send({class="st">"type": class="st">"http.response.start", class="st">"status": self.status_code, class="st">"headers": self.raw_headers})
        async for chunk in self.body_iterator:
            if not isinstance(chunk, bytes | memoryview):
                chunk = chunk.encode(self.charset)
            await send({class="st">"type": class="st">"http.response.body", class="st">"body": chunk, class="st">"more_body": True})

        await send({class="st">"type": class="st">"http.response.body", class="st">"body": bclass="st">"", class="st">"more_body": False})

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope[class="st">"type"] == class="st">"websocket":
            send = self._wrap_websocket_denial_send(send)
            await self.stream_response(send)
            if self.background is not None:
                await self.background()
            return

        spec_version = tuple(map(int, scope.get(class="st">"asgi", {}).get(class="st">"spec_version", class="st">"2.0").split(class="st">".")))

        if spec_version >= (2, 4):
            try:
                await self.stream_response(send)
            except OSError:
                raise ClientDisconnect()
        else:
            async with create_collapsing_task_group() as task_group:

                async def wrap(func: Callable[[], Awaitable[None]]) -> None:
                    await func()
                    task_group.cancel_scope.cancel()

                task_group.start_soon(wrap, partial(self.stream_response, send))