feat: Add EventSourceResponse with FastAPI-style SSE support (#203)

- EventSourceResponse with auto SSE framing, compression skip, keep-alive pings
- ServerSentEvent struct for full control over event/id/retry/comment fields
- response_class parameter on all HTTP method decorators
- Rust: zero-copy chunks via PyBackedBytes, PyBackedStr for media_type
- Rust: always skip compression for text/event-stream streams
- Rust: keepalive ping stream wrapper using tokio::time::timeout
- Dedicated SSE documentation page

fix: Resolve task affinity bug in async streaming (fixes opentelemetry/anyio errors)

Rewrote async stream forwarding to schedule the Python generator iteration
on the same asyncio event loop task that created the generator context.
The previous approach used tokio::spawn + join_all which drove the generator
from a different task, breaking anyio cancel scopes and opentelemetry
context propagation (ContextVar tokens created in one task couldn't be
detached in another).

Author: FarhanAliRaza

docs/src/ref/responses.md

@@ -173,9 +173,97 @@ BOLT_ALLOWED_FILE_PATHS = [
 ]
 ```
 
+## EventSourceResponse
+
+Server-Sent Events response with automatic SSE framing, compression skipping, and keep-alive pings. See the [SSE topic guide](../topics/sse.md) for full documentation.
+
+### Implicit pattern
+
+```python
+from collections.abc import AsyncIterable
+from django_bolt.responses import EventSourceResponse
+
+@api.get("/events", response_class=EventSourceResponse)
+async def events() -> AsyncIterable[dict]:
+    for i in range(10):
+        yield {"count": i}
+        await asyncio.sleep(1)
+```
+
+### Explicit pattern
+
+```python
+from django_bolt.responses import EventSourceResponse
+
+@api.get("/events")
+async def events():
+    async def generate():
+        yield {"count": i}
+    return EventSourceResponse(generate())
+```
+
+### Parameters
+
+| Parameter | Type | Default | Description |
+|-----------|------|---------|-------------|
+| `content` | generator | required | Generator instance |
+| `status_code` | `int` | `200` | HTTP status code |
+| `headers` | `dict` | `None` | Additional response headers |
+| `ping_interval` | `float \| None` | `15.0` | Keep-alive ping interval in seconds (`None` to disable) |
+
+### Automatic behavior
+
+- **SSE framing**: Yielded objects are JSON-serialized and wrapped in `data: ...\n\n`
+- **Compression**: Automatically skipped (no `@no_compress` needed)
+- **Headers**: `Cache-Control`, `X-Accel-Buffering`, `Content-Type` set automatically
+- **Keep-alive**: `: ping` comments sent when idle to prevent proxy timeouts
+
+## ServerSentEvent
+
+Represents a single SSE event with full control over all fields.
+
+```python
+from django_bolt.responses import ServerSentEvent
+
+yield ServerSentEvent(data={"count": 1}, event="update", id="1")
+yield ServerSentEvent(raw_data="plain text", comment="keepalive")
+yield ServerSentEvent(retry=5000)
+```
+
+### Fields
+
+| Field | Type | Default | Description |
+|-------|------|---------|-------------|
+| `data` | `Any` | `None` | JSON-serialized event payload |
+| `raw_data` | `str \| None` | `None` | Raw string payload (mutually exclusive with `data`) |
+| `event` | `str \| None` | `None` | Event type name |
+| `id` | `str \| None` | `None` | Event ID (no null characters) |
+| `retry` | `int \| None` | `None` | Reconnection time in ms (non-negative) |
+| `comment` | `str \| None` | `None` | Comment line (`: ` prefix) |
+
+## format_sse_event
+
+Build SSE wire-format bytes from pre-serialized data.
+
+```python
+from django_bolt.responses import format_sse_event
+
+# Simple data event
+frame = format_sse_event(data_str='{"count": 1}')
+# b'data: {"count": 1}\n\n'
+
+# Full event
+frame = format_sse_event(data_str="payload", event="update", id="42", retry=5000)
+# b'event: update\ndata: payload\nid: 42\nretry: 5000\n\n'
+
+# Pre-encoded bytes (avoids decode/encode round-trip)
+frame = format_sse_event(data_bytes=b'{"count": 1}')
+# b'data: {"count": 1}\n\n'
+```
+
 ## StreamingResponse
 
-Streaming response for generators.
+General-purpose streaming response for generators.
 
 ```python
 from django_bolt import StreamingResponse
@@ -193,9 +281,9 @@ return StreamingResponse(generate(), media_type="text/plain")
 async def async_generate():
     for i in range(100):
         await asyncio.sleep(0.1)
-        yield f"data: {i}\n\n"
+        yield f"chunk {i}\n"
 
-return StreamingResponse(async_generate(), media_type="text/event-stream")
+return StreamingResponse(async_generate(), media_type="text/plain")
 ```
 
 ### Parameters
@@ -212,7 +300,7 @@ return StreamingResponse(async_generate(), media_type="text/event-stream")
 | Type | Use case |
 |------|----------|
 | `text/plain` | Plain text streaming |
-| `text/event-stream` | Server-Sent Events (SSE) |
+| `text/event-stream` | Server-Sent Events (prefer `EventSourceResponse`) |
 | `application/octet-stream` | Binary data |
 | `application/json` | JSON streaming (NDJSON) |
 

docs/src/topics/middleware.md

@@ -130,6 +130,10 @@ async def stream():
     return StreamingResponse(generate())
 ```
 
+!!! note
+    `EventSourceResponse` and all SSE streams (`text/event-stream`) skip compression
+    automatically. You only need `@no_compress` for non-SSE streaming.
+
 ### Compression settings
 
 Configure in `settings.py`:

docs/src/topics/responses.md

@@ -250,149 +250,17 @@ async def async_stream():
 
 ### Server-Sent Events (SSE)
 
-Create SSE endpoints for real-time updates. SSE is a standard for pushing events from server to browser over HTTP.
-
-#### Basic SSE
+For SSE endpoints, see the dedicated [Server-Sent Events guide](sse.md). Django-Bolt provides `EventSourceResponse` with automatic SSE framing, compression skipping, and keep-alive pings:
 
 ```python
-import asyncio
-
-@api.get("/events")
-async def events():
-    async def generate():
-        for i in range(10):
-            yield f"data: message-{i}\n\n"
-            await asyncio.sleep(1)
-
-    return StreamingResponse(generate(), media_type="text/event-stream")
-```
-
-#### SSE format
-
-Each event is terminated by a double newline (`\n\n`). Fields:
-
-| Field | Description |
-|-------|-------------|
-| `data:` | Event data (required) |
-| `event:` | Event type (optional, default: "message") |
-| `id:` | Event ID for reconnection (optional) |
-| `retry:` | Reconnection time in ms (optional) |
-
-#### Full SSE event format
-
-```python
-@api.get("/sse-events")
-async def sse_events():
-    async def generate():
-        for i in range(5):
-            # Full SSE event with all fields
-            yield f"event: update\nid: {i}\ndata: {{\"count\": {i}}}\n\n"
-            await asyncio.sleep(0.5)
-
-    return StreamingResponse(generate(), media_type="text/event-stream")
-```
-
-Client receives:
-```
-event: update
-id: 0
-data: {"count": 0}
-
-event: update
-id: 1
-data: {"count": 1}
-```
-
-#### Sync generators for SSE
-
-You can use sync generators for CPU-bound operations:
-
-```python
-import time
-
-@api.get("/sync-sse")
-async def sync_sse():
-    def generate():
-        for i in range(5):
-            yield f"data: sync-message-{i}\n\n"
-            time.sleep(0.1)
-
-    return StreamingResponse(generate(), media_type="text/event-stream")
-```
-
-#### Mixed data types
-
-Generators can yield both strings and bytes:
+from collections.abc import AsyncIterable
+from django_bolt.responses import EventSourceResponse
 
-```python
-@api.get("/mixed-sse")
-async def mixed_sse():
-    async def generate():
-        yield "data: string message\n\n"
-        yield b"data: bytes message\n\n"  # Also works
-
-    return StreamingResponse(generate(), media_type="text/event-stream")
-```
-
-#### SSE with cleanup
-
-Use try/finally for resource cleanup when clients disconnect:
-
-```python
-@api.get("/sse-with-cleanup")
-async def sse_with_cleanup():
-    async def generate():
-        try:
-            yield "data: START\n\n"
-            for i in range(100):
-                yield f"data: chunk-{i}\n\n"
-                await asyncio.sleep(0.1)
-            yield "data: END\n\n"
-        finally:
-            # This runs when client disconnects
-            print("Client disconnected, cleaning up")
-
-    return StreamingResponse(generate(), media_type="text/event-stream")
-```
-
-#### SSE headers
-
-SSE endpoints should include these headers for proper behavior:
-
-```python
-@api.get("/sse")
-async def sse():
-    async def generate():
-        for i in range(10):
-            yield f"data: {i}\n\n"
-            await asyncio.sleep(1)
-
-    return StreamingResponse(
-        generate(),
-        media_type="text/event-stream",
-        headers={
-            "Cache-Control": "no-cache",
-            "X-Accel-Buffering": "no",  # Disable nginx buffering
-        }
-    )
-```
-
-### Disabling compression for streams
-
-Streaming responses should not be compressed. Use `@no_compress`:
-
-```python
-from django_bolt.middleware import no_compress
-
-@api.get("/sse")
-@no_compress
-async def sse():
-    async def generate():
-        for i in range(10):
-            yield f"data: {i}\n\n"
-            await asyncio.sleep(1)
-
-    return StreamingResponse(generate(), media_type="text/event-stream")
+@api.get("/events", response_class=EventSourceResponse)
+async def events() -> AsyncIterable[dict]:
+    for i in range(10):
+        yield {"count": i}
+        await asyncio.sleep(1)
 ```
 
 ## Response with custom headers