src/genro_asgi/streaming.py¶
Source from this local checkout, regenerated when the reader rebuilds.
Line links use #L<number>; a GitHub line range opens its first line.
1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 # https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """Streaming HTTP response: a chunked ASGI sibling of ``Response``.16 17 ``Response`` (response.py) is flat and BUFFERED — two ASGI messages, the whole18 body in memory. A stream is a different shape, not a variant, so it is a19 separate slotted class rather than a subclass: ``StreamingResponse`` sends the20 ``http.response.start`` once, then one ``http.response.body`` per chunk with21 ``more_body=True``, and a terminal empty body with ``more_body=False``. It has22 NO ``set_result`` (the buffered type-dispatch is deliberately not carried over):23 the body is an async iterator of ``bytes`` the caller supplies.24 25 The iterator is the whole contract — a plain ``async for`` over user chunks, an26 ``SseStream`` (sse.py), or any bounded event source. This class only frames the27 ASGI message sequence around it; backpressure and heartbeats live in the source.28 """29 30 from __future__ import annotations31 32 from collections.abc import AsyncIterable, Mapping33 34 from .types import Receive, Scope, Send35 36 __all__ = ["StreamingResponse"]37 38 HeadersInput = Mapping[str, str] | list[tuple[str, str]] | None39 40 41 class StreamingResponse:42 """Chunked HTTP response, usable directly as an ASGI application.43 44 Example:45 >>> async def chunks():46 ... yield b"one"47 ... yield b"two"48 >>> response = StreamingResponse(chunks(), media_type="text/plain")49 >>> await response(scope, receive, send)50 """51 52 __slots__ = ("body_iterator", "status_code", "media_type", "_headers")53 54 charset: str = "utf-8"55 56 def __init__(57 self,58 body_iterator: AsyncIterable[bytes],59 status_code: int = 200,60 headers: HeadersInput = None,61 media_type: str | None = None,62 ) -> None:63 """Build a streaming response over an async iterator of byte chunks."""64 self.body_iterator = body_iterator65 self.status_code = status_code66 self.media_type = media_type67 if headers is None:68 self._headers = []69 elif isinstance(headers, list):70 self._headers = list(headers)71 else:72 self._headers = list(headers.items())73 if media_type is not None:74 names = {name.lower() for name, _ in self._headers}75 if "content-type" not in names:76 self._headers.append(("content-type", self._content_type(media_type)))77 78 def _content_type(self, media_type: str) -> str:79 """Content-Type value, appending the charset for text types lacking one."""80 if media_type.startswith("text/") and "charset" not in media_type:81 return f"{media_type}; charset={self.charset}"82 return media_type83 84 def _build_headers(self) -> list[tuple[bytes, bytes]]:85 """ASGI headers: names lowercased, latin-1 encoded (HTTP standard)."""86 return [87 (name.lower().encode("latin-1"), value.encode("latin-1"))88 for name, value in self._headers89 ]90 91 def set_header(self, name: str, value: str) -> None:92 """Append a response header (before the response is sent)."""93 self._headers.append((name, value))94 95 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:96 """ASGI interface: ``http.response.start`` then one body per chunk.97 98 Each chunk goes out with ``more_body=True``; a terminal empty body with99 ``more_body=False`` closes the stream (never omitted, even when the100 iterator yields nothing).101 """102 await send(103 {104 "type": "http.response.start",105 "status": self.status_code,106 "headers": self._build_headers(),107 }108 )109 async for chunk in self.body_iterator:110 await send({"type": "http.response.body", "body": chunk, "more_body": True})111 await send({"type": "http.response.body", "body": b"", "more_body": False})