fix

docs/feature.md

@@ -23,7 +23,7 @@
 | Тема | Зазор | Комментарий |
 |------|--------|-------------|
 | **WebSockets** | Нет | RSGI/текущий стек заточен под HTTP-запрос/ответ. Нужна отдельная ветка согласно Granian/RSGI для WS и мост в Python. |
-| **SSE / длинный стрим ответа** | Нет / ограничено | Сейчас ответы через `response_*` Granian; потоковая отдача чанками — отдельный дизайн. |
+| **SSE / длинный стрим ответа** | Частично | Есть `send_sse` (см. [sse.md](sse.md)); инкрементальный стрим зависит от `response_stream` в серверном протоколе, иначе fallback в буферизованный ответ. |
 | **HTTP/2 push, trailers** | Не в фокусе | Обычно на стороне сервера; фреймворк редко экспонирует. |
 | **ASGI: lifespan, `websocket`, background** | Частично | Мост ASGI — **только `http`**; нет полноценного ASGI-приложения с lifecycle из спеки. |
 

docs/handlers.md

@@ -20,6 +20,7 @@ handler(**kwargs)
 | `form` | `read_form_body` is true on **POST**, **PUT**, **PATCH**, or **DELETE** | `dict` of string keys to string values (same last-wins semantics as `query`). Parsed from `application/x-www-form-urlencoded` or non-file parts of `multipart/form-data` |
 | `files` | `read_form_body` is true and the request is `multipart/form-data` | `list` of `dict`s with `name`, `filename` (`str` or missing), `content_type`, and `data` (`bytes`). In-memory only; there is no streaming spool to disk in the current implementation |
 | `body` | Raw body bytes when JSON and form modes are not used | `bytes`. **Not** passed when `read_form_body` is enabled (use `form` / `files` instead) |
+| `protocol` | Handler declares a `protocol` parameter (or `**kwargs`) | Underlying RSGI protocol object for advanced response flows (e.g. SSE via `send_sse`) |
 | `claims` | `require_jwt` is true and the JWT validates | The decoded JSON claims as a Python object (typically a `dict`) |
 | Named dependencies | `dependencies=[("name", factory), ...]` | Return value of each factory, in order (see [dependencies.md](dependencies.md)). Only dependencies whose **names** appear on the route handler’s signature (or `**kwargs`) are passed to the handler—intermediate-only dependencies are not forwarded |
 
@@ -97,3 +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)

docs/index.md

@@ -45,6 +45,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 |
 | [HTTP/2 with Granian](http2.md) | Transport guarantees vs server/proxy responsibilities |
 | [Dependencies](dependencies.md) | `Depends`, `dependencies=[...]`, `freeze` |
 | [OpenAPI](openapi.md) | `openapi.json` route, title, `openapi_json()` |

docs/sse.md

@@ -0,0 +1,44 @@
+# 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: <value>`),
+  - `SSEEvent(data=..., event=..., id=..., retry=...)`.
+
+## Streaming behavior
+
+- On RSGI protocols exposing `response_stream`, chunks are written incrementally.
+- On transports without streaming support (e.g. current ASGI bridge test path), 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)

oxyroute/__init__.py

@@ -8,6 +8,7 @@
 from oxyroute.response import Response
 from oxyroute.router import APIRouter
 from oxyroute.security_headers import SecurityHeadersConfig
+from oxyroute.sse import SSEEvent, send_sse, sse_done
 
 __all__ = [
     "APIRouter",
@@ -17,11 +18,14 @@
     "Depends",
     "HTTPException",
     "Response",
+    "SSEEvent",
     "SecurityHeadersConfig",
     "__version__",
     "apply_cors",
     "apply_csrf",
     "csrf_layer",
     "decode_jwt_hs",
+    "send_sse",
+    "sse_done",
 ]
 __version__ = "0.1.0"

oxyroute/sse.py

@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+from collections.abc import AsyncIterable, Iterable
+from dataclasses import dataclass
+from typing import Any
+
+__all__ = ["SSEEvent", "send_sse", "sse_done"]
+
+
+@dataclass(slots=True)
+class SSEEvent:
+    data: str
+    event: str | None = None
+    id: str | None = None
+    retry: int | None = None
+
+
+class _SSEDone:
+    __slots__ = ()
+    __oxyroute_stream_done__ = True
+
+
+def sse_done() -> Any:
+    """Return a marker value telling OxyRoute the response was already sent."""
+    return _SSEDone()
+
+
+def _format_sse_event(ev: SSEEvent) -> str:
+    lines: list[str] = []
+    if ev.event:
+        lines.append(f"event: {ev.event}")
+    if ev.id:
+        lines.append(f"id: {ev.id}")
+    if ev.retry is not None:
+        lines.append(f"retry: {ev.retry}")
+    for line in ev.data.splitlines() or [""]:
+        lines.append(f"data: {line}")
+    return "\n".join(lines) + "\n\n"
+
+
+def _to_sse_chunk(item: Any) -> str:
+    if isinstance(item, SSEEvent):
+        return _format_sse_event(item)
+    if isinstance(item, str):
+        return _format_sse_event(SSEEvent(data=item))
+    raise TypeError("SSE items must be str or SSEEvent")
+
+
+async def send_sse(
+    protocol: Any,
+    events: Iterable[str | SSEEvent] | AsyncIterable[str | SSEEvent],
+    *,
+    status: int = 200,
+    headers: list[tuple[str, str]] | None = None,
+) -> Any:
+    """
+    Send SSE response 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", "text/event-stream; charset=utf-8"),
+        ("cache-control", "no-cache"),
+        ("connection", "keep-alive"),
+    ]
+    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(events, "__aiter__"):
+            async for item in events:  # type: ignore[union-attr]
+                await stream.send_str(_to_sse_chunk(item))
+        else:
+            for item in events:  # type: ignore[not-an-iterable]
+                await stream.send_str(_to_sse_chunk(item))
+        return sse_done()
+
+    chunks: list[str] = []
+    if hasattr(events, "__aiter__"):
+        async for item in events:  # type: ignore[union-attr]
+            chunks.append(_to_sse_chunk(item))
+    else:
+        for item in events:  # type: ignore[not-an-iterable]
+            chunks.append(_to_sse_chunk(item))
+    protocol.response_str(status, base_headers, "".join(chunks))
+    return sse_done()

src/dispatch.rs

@@ -672,6 +672,9 @@ pub async fn run_rsgi(
         } else if !body_bytes.is_empty() && body_json.is_none() {
             kwargs.set_item("body", PyBytes::new(py, &body_bytes))?;
         }
+        if handler_varkw || handler_param_names.contains("protocol") {
+            kwargs.set_item("protocol", protocol.bind(py))?;
+        }
         let res = handler.bind(py).call((), Some(&kwargs))?.unbind();
         Ok((res, is_async))
     }) {
@@ -727,6 +730,12 @@ pub async fn run_rsgi(
 /// Return value of a user handler, mapped to an HTTP body and headers.
 fn map_handler_return(py: Python<'_>, out: &Py<PyAny>) -> PyResult<HandlerMap> {
     let b = out.bind(py);
+    if b.getattr("__oxyroute_stream_done__")
+        .and_then(|x| x.extract::<bool>())
+        .unwrap_or(false)
+    {
+        return Ok(HandlerMap::AlreadySent);
+    }
     // Before `extract::<String>`: some non-`str` objects may still coerce in edge cases;
     // `Response` must be recognized first.
     if is_oxyroute_response(py, b)? {
@@ -791,6 +800,7 @@ fn map_handler_return(py: Python<'_>, out: &Py<PyAny>) -> PyResult<HandlerMap> {
 }
 
 enum HandlerMap {
+    AlreadySent,
     WithHeaders {
         status: u16,
         body: Vec<u8>,
@@ -808,8 +818,12 @@ async fn send_handler_map(
     is_head: bool,
     mapped: HandlerMap,
 ) -> PyResult<PyObject> {
+    if matches!(mapped, HandlerMap::AlreadySent) {
+        return Ok(Python::with_gil(|py| py.None()));
+    }
     if is_head {
         match mapped {
+            HandlerMap::AlreadySent => Ok(Python::with_gil(|py| py.None())),
             HandlerMap::WithHeaders {
                 status,
                 body,
@@ -823,6 +837,7 @@ async fn send_handler_map(
         }
     } else {
         match mapped {
+            HandlerMap::AlreadySent => Ok(Python::with_gil(|py| py.None())),
             HandlerMap::WithHeaders {
                 status,
                 body,
@@ -867,6 +882,7 @@ fn merge_header_pairs_replace(mapped: HandlerMap, extra: &[(String, String)]) ->
         return mapped;
     }
     match mapped {
+        HandlerMap::AlreadySent => HandlerMap::AlreadySent,
         HandlerMap::WithHeaders {
             status,
             body,
@@ -906,6 +922,7 @@ fn merge_header_pairs_if_absent(mapped: HandlerMap, extra: &[(String, String)])
         return mapped;
     }
     match mapped {
+        HandlerMap::AlreadySent => HandlerMap::AlreadySent,
         HandlerMap::WithHeaders {
             status,
             body,

tests/test_sse.py

@@ -0,0 +1,28 @@
+"""SSE helper over RSGI protocol (issue #51)."""
+
+from __future__ import annotations
+
+import asyncio
+
+import httpx
+from oxyroute import App, send_sse
+
+
+def test_sse_response_body_and_content_type() -> None:
+    app = App()
+
+    @app.get("/events")
+    async def events(protocol: object) -> object:
+        data = ["ready", "tick"]
+        return await send_sse(protocol, data)  # type: ignore[arg-type]
+
+    async def _run() -> None:
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
+            r = await c.get("/events")
+        assert r.status_code == 200
+        assert "text/event-stream" in r.headers.get("content-type", "")
+        assert "data: ready\n\n" in r.text
+        assert "data: tick\n\n" in r.text
+
+    asyncio.run(_run())