feat(streaming): generic stream responses (generators) for non-SSE chunks

docs/feature.md

@@ -27,7 +27,7 @@
 | Тема | Зазор | Комментарий |
 |------|--------|-------------|
 | **WebSockets** | Реализованы | Native RSGI WebSocket: `@app.websocket(path)`, `oxyroute.WebSocket`; см. [websocket.md](websocket.md). Нет high-level subprotocol API. |
-| **SSE / длинный стрим ответа** | Частично | Есть `send_sse` (см. [sse.md](sse.md)); инкрементальный стрим использует `response_stream` Granian RSGI. |
+| **SSE / длинный стрим ответа** | Частично | Есть `send_sse` (см. [streaming.md](streaming.md)); инкрементальный стрим использует `response_stream` Granian RSGI. |
 | **HTTP/2 push, trailers** | Не в фокусе | Обычно на стороне сервера; фреймворк редко экспонирует. |
 | **ASGI совместимость** | Удалена в v0.3.0 | Поддерживается только RSGI (Granian `--interface rsgi`). |
 

docs/handlers.md

@@ -98,4 +98,4 @@ For a configurable **`allow_origins` / `allow_methods` / `allow_headers`** flow
 - [CORS](cors.md)
 - [Security headers](security-headers.md)
 - [CSRF](csrf.md)
-- [SSE](sse.md)
+- [SSE](streaming.md)

docs/index.md

@@ -46,7 +46,7 @@ Granian still invokes a Python `App` object; the “win” is doing routing, bod
 | [Security headers](security-headers.md) | `SecurityHeadersConfig`, HSTS, CSP |
 | [CSRF](csrf.md) | Double-submit, `apply_csrf`, `csrf_layer` + CORS |
 | [JWT](jwt.md) | `require_jwt`, HS* / RSA / EC PEM, `decode_jwt_hs` (HS* tests) |
-| [SSE](sse.md) | `send_sse`, event framing, streaming caveats |
+| [Streaming & SSE](streaming.md) | `stream_bytes`, `stream_text`, `send_sse`, streaming caveats |
 | [WebSockets](websocket.md) | Native RSGI `@app.websocket(path)` and `oxyroute.WebSocket` |
 | [HTTP/2 with Granian](http2.md) | Transport guarantees vs server/proxy responsibilities |
 | [Dependencies](dependencies.md) | `Depends`, `dependencies=[...]`, `freeze` |

docs/streaming.md

@@ -0,0 +1,56 @@
+# Streaming Responses
+
+[← Documentation index](index.md)
+
+OxyRoute provides streaming response helpers in `oxyroute.streaming` for returning chunked HTTP responses without buffering the entire body in memory. Server-Sent Events (SSE) are available in `oxyroute.sse`.
+
+## Quick start
+
+```python
+import asyncio
+from oxyroute import App, stream_text
+
+app = App()
+
+@app.get("/logs")
+async def logs(protocol):
+    async def tail_logs():
+        for i in range(5):
+            yield f"Log line {i}\n"
+            await asyncio.sleep(1)
+
+    # `stream_text` formats chunks as `text/plain`
+    return await stream_text(protocol, tail_logs())
+```
+
+## Available Helpers
+
+All helpers require the `protocol` argument and an iterable or async iterable of data.
+
+- `stream_bytes(protocol, iterable, *, status=200, headers=None, content_type="application/octet-stream")`
+  Streams raw `bytes`. Useful for file downloads and proxying binary streams.
+
+- `stream_text(protocol, iterable, *, status=200, headers=None, content_type="text/plain; charset=utf-8")`
+  Streams `str` chunks.
+
+- `stream_jsonl(protocol, iterable, *, status=200, headers=None)`
+  Takes an iterable of dicts/lists/objects and streams them as NDJSON (JSON-Lines) with `content-type: application/x-ndjson; charset=utf-8`.
+
+- `send_sse(protocol, events, *, status=200, headers=None)`
+  Streams items as Server-Sent Events with `content-type: text/event-stream; charset=utf-8`.
+
+## Behavior & Backpressure
+
+- On RSGI servers supporting `response_stream` (like Granian), chunks are sent incrementally.
+- Awaiting the streaming helpers automatically respects TCP backpressure. If the client is slow to read, Granian pauses the underlying stream, causing `await stream.send_bytes(...)` to block, which in turn pauses your async generator.
+- On transports without streaming support (e.g., the integrated test client or ASGI bridging), OxyRoute falls back to buffering all chunks into memory and returning a single response.
+
+## Caveats
+
+- Browser or intermediate proxy buffering can delay chunk delivery unless buffering is explicitly disabled at the edge.
+- Handlers returning streams **must** be `async def` and must `await` the streaming helper, because the helpers themselves run async I/O.
+
+## See also
+
+- [Handlers](handlers.md)
+- [HTTP/2 with Granian](http2.md)

docs/usage.md

@@ -465,7 +465,7 @@ async def events(protocol):
     return await send_sse(protocol, [SSEEvent(data="hello")])
 ```
 
-See [sse.md](sse.md) for details and caveats.
+See [streaming.md](streaming.md) for details and caveats.
 
 ## OpenAPI
 

oxyroute/__init__.py

@@ -10,6 +10,7 @@
 from oxyroute.router import APIRouter
 from oxyroute.security_headers import SecurityHeadersConfig
 from oxyroute.sse import SSEEvent, send_sse, sse_done
+from oxyroute.streaming import stream_bytes, stream_jsonl, stream_text
 
 __all__ = [
     "APIRouter",
@@ -31,5 +32,8 @@
     "decode_jwt_hs",
     "send_sse",
     "sse_done",
+    "stream_bytes",
+    "stream_jsonl",
+    "stream_text",
 ]
 __version__ = "0.4.0"

oxyroute/streaming.py

@@ -0,0 +1,126 @@
+from __future__ import annotations
+
+import json
+from collections.abc import AsyncIterable, Iterable
+from typing import Any
+
+__all__ = ["stream_bytes", "stream_done", "stream_jsonl", "stream_text"]
+
+
+class _StreamDone:
+    __slots__ = ()
+    __oxyroute_stream_done__ = True
+
+
+def stream_done() -> Any:
+    """Return a marker value telling OxyRoute the response was already sent."""
+    return _StreamDone()
+
+
+async def stream_bytes(
+    protocol: Any,
+    iterable: Iterable[bytes] | AsyncIterable[bytes],
+    *,
+    status: int = 200,
+    headers: list[tuple[str, str]] | None = None,
+    content_type: str = "application/octet-stream",
+) -> Any:
+    """
+    Stream raw bytes via RSGI protocol.
+
+    If `response_stream` is available (Granian RSGI), writes chunks incrementally.
+    Otherwise (ASGI bridge/test transports), falls back to a single response body.
+    """
+    base_headers: list[tuple[str, str]] = [("content-type", content_type)]
+    if headers:
+        base_headers.extend(headers)
+
+    stream_factory = getattr(protocol, "response_stream", None)
+    if callable(stream_factory):
+        stream = stream_factory(status, base_headers)
+        if hasattr(iterable, "__aiter__"):
+            async for chunk in iterable:  # type: ignore[union-attr]
+                await stream.send_bytes(chunk)
+        else:
+            for chunk in iterable:  # type: ignore[not-an-iterable]
+                await stream.send_bytes(chunk)
+        return stream_done()
+
+    # Fallback for test/ASGI transports
+    chunks: list[bytes] = []
+    if hasattr(iterable, "__aiter__"):
+        async for chunk in iterable:  # type: ignore[union-attr]
+            chunks.append(chunk)
+    else:
+        for chunk in iterable:  # type: ignore[not-an-iterable]
+            chunks.append(chunk)
+    protocol.response_bytes(status, base_headers, b"".join(chunks))
+    return stream_done()
+
+
+async def stream_text(
+    protocol: Any,
+    iterable: Iterable[str] | AsyncIterable[str],
+    *,
+    status: int = 200,
+    headers: list[tuple[str, str]] | None = None,
+    content_type: str = "text/plain; charset=utf-8",
+) -> Any:
+    """
+    Stream text via RSGI protocol.
+
+    If `response_stream` is available (Granian RSGI), writes chunks incrementally.
+    Otherwise (ASGI bridge/test transports), falls back to a single response body.
+    """
+    base_headers: list[tuple[str, str]] = [("content-type", content_type)]
+    if headers:
+        base_headers.extend(headers)
+
+    stream_factory = getattr(protocol, "response_stream", None)
+    if callable(stream_factory):
+        stream = stream_factory(status, base_headers)
+        if hasattr(iterable, "__aiter__"):
+            async for chunk in iterable:  # type: ignore[union-attr]
+                await stream.send_str(chunk)
+        else:
+            for chunk in iterable:  # type: ignore[not-an-iterable]
+                await stream.send_str(chunk)
+        return stream_done()
+
+    chunks: list[str] = []
+    if hasattr(iterable, "__aiter__"):
+        async for chunk in iterable:  # type: ignore[union-attr]
+            chunks.append(chunk)
+    else:
+        for chunk in iterable:  # type: ignore[not-an-iterable]
+            chunks.append(chunk)
+    protocol.response_str(status, base_headers, "".join(chunks))
+    return stream_done()
+
+
+async def stream_jsonl(
+    protocol: Any,
+    iterable: Iterable[Any] | AsyncIterable[Any],
+    *,
+    status: int = 200,
+    headers: list[tuple[str, str]] | None = None,
+) -> Any:
+    """
+    Stream NDJSON (JSON-Lines) via RSGI protocol.
+    """
+
+    async def _jsonl_iter() -> AsyncIterable[str]:
+        if hasattr(iterable, "__aiter__"):
+            async for item in iterable:  # type: ignore[union-attr]
+                yield json.dumps(item) + "\n"
+        else:
+            for item in iterable:  # type: ignore[not-an-iterable]
+                yield json.dumps(item) + "\n"
+
+    return await stream_text(
+        protocol,
+        _jsonl_iter(),
+        status=status,
+        headers=headers,
+        content_type="application/x-ndjson; charset=utf-8",
+    )

tests/test_streaming.py

@@ -0,0 +1,76 @@
+from oxyroute import App, stream_bytes, stream_jsonl, stream_text
+from oxyroute.testing import TestClient
+
+
+def test_stream_text():
+    app = App()
+
+    @app.get("/text")
+    async def text_handler(protocol):
+        def _iter():
+            yield "hello"
+            yield " "
+            yield "world"
+
+        return await stream_text(protocol, _iter())
+
+    with TestClient(app) as client:
+        resp = client.get("/text")
+        assert resp.status_code == 200
+        assert resp.text == "hello world"
+        assert resp.headers["content-type"] == "text/plain; charset=utf-8"
+
+
+def test_stream_bytes():
+    app = App()
+
+    @app.get("/bytes")
+    async def bytes_handler(protocol):
+        def _iter():
+            yield b"123"
+            yield b"456"
+
+        return await stream_bytes(protocol, _iter(), status=201, headers=[("x-custom", "foo")])
+
+    with TestClient(app) as client:
+        resp = client.get("/bytes")
+        assert resp.status_code == 201
+        assert resp.content == b"123456"
+        assert resp.headers["content-type"] == "application/octet-stream"
+        assert resp.headers["x-custom"] == "foo"
+
+
+def test_stream_jsonl():
+    app = App()
+
+    @app.get("/jsonl")
+    async def jsonl_handler(protocol):
+        def _iter():
+            yield {"id": 1, "name": "foo"}
+            yield {"id": 2, "name": "bar"}
+
+        return await stream_jsonl(protocol, _iter())
+
+    with TestClient(app) as client:
+        resp = client.get("/jsonl")
+        assert resp.status_code == 200
+        assert resp.text == '{"id": 1, "name": "foo"}\n{"id": 2, "name": "bar"}\n'
+        assert resp.headers["content-type"] == "application/x-ndjson; charset=utf-8"
+
+
+def test_stream_async_iter():
+    app = App()
+
+    @app.get("/async")
+    async def async_handler(protocol):
+        async def _iter():
+            yield "async"
+            yield " "
+            yield "iter"
+
+        return await stream_text(protocol, _iter())
+
+    with TestClient(app) as client:
+        resp = client.get("/async")
+        assert resp.status_code == 200
+        assert resp.text == "async iter"