diff --git a/docs/feature.md b/docs/feature.md index a04304e..0099abb 100644 --- a/docs/feature.md +++ b/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`). | diff --git a/docs/handlers.md b/docs/handlers.md index f2fdda8..50a6838 100644 --- a/docs/handlers.md +++ b/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) diff --git a/docs/index.md b/docs/index.md index e701419..9c9af3f 100644 --- a/docs/index.md +++ b/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` | diff --git a/docs/sse.md b/docs/sse.md deleted file mode 100644 index ab4caeb..0000000 --- a/docs/sse.md +++ /dev/null @@ -1,44 +0,0 @@ -# Server-Sent Events (SSE) - -[← Documentation index](index.md) - -OxyRoute provides a small SSE helper in `oxyroute.sse` for HTTP event streams. - -## Quick start - -```python -from oxyroute import App, send_sse - -app = App() - - -@app.get("/events") -async def events(protocol): - return await send_sse(protocol, ["ready", "tick"]) -``` - -## API - -- `send_sse(protocol, events, *, status=200, headers=None)`: - - sets `content-type: text/event-stream; charset=utf-8`, - - formats items as SSE frames (`data: ...\n\n`), - - returns a sentinel consumed by OxyRoute so no second response is emitted. -- Event items can be: - - `str` (serialized as `data: `), - - `SSEEvent(data=..., event=..., id=..., retry=...)`. - -## Streaming behavior - -- On RSGI protocols exposing `response_stream` (Granian RSGI), chunks are written incrementally. -- On transports without streaming support (e.g. the in-process httpx test transport), OxyRoute falls back to one buffered `response_str` body with SSE framing. - -## Caveats - -- Browser/proxy buffering can delay event delivery unless buffering is disabled at the edge. -- SSE is one-way server-to-client messaging over HTTP; use WebSockets for bi-directional flows. -- HTTP/1.1 and HTTP/2 transport negotiation is server/proxy responsibility; SSE framing itself is unchanged. - -## See also - -- [Handlers](handlers.md) -- [HTTP/2 with Granian](http2.md) diff --git a/docs/streaming.md b/docs/streaming.md new file mode 100644 index 0000000..542a74a --- /dev/null +++ b/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) diff --git a/docs/usage.md b/docs/usage.md index 567d756..9d0abb1 100644 --- a/docs/usage.md +++ b/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 diff --git a/oxyroute/__init__.py b/oxyroute/__init__.py index 0c11df6..8391f3c 100644 --- a/oxyroute/__init__.py +++ b/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" diff --git a/oxyroute/streaming.py b/oxyroute/streaming.py new file mode 100644 index 0000000..85af591 --- /dev/null +++ b/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", + ) diff --git a/tests/test_streaming.py b/tests/test_streaming.py new file mode 100644 index 0000000..508a54a --- /dev/null +++ b/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"