diff --git a/docs/en/tutorial/index.md b/docs/en/tutorial/index.md index 3d39d1f..f83084f 100644 --- a/docs/en/tutorial/index.md +++ b/docs/en/tutorial/index.md @@ -38,6 +38,8 @@ The tutorial is divided into several sections: - [Middleware](middleware/index.md) - Global request logic - [Security](security/index.md) - Built-in protection - [GraphQL](graphql/index.md) - GraphQL support +- [SSE](sse/index.md) - Server-Sent Events +- [WebSocket](websocket/index.md) - WebSocket connections - [OpenAPI](openapi/index.md) - Swagger UI ## How to Use diff --git a/docs/en/tutorial/sse/index.md b/docs/en/tutorial/sse/index.md new file mode 100644 index 0000000..68ebe6d --- /dev/null +++ b/docs/en/tutorial/sse/index.md @@ -0,0 +1,183 @@ +# Server-Sent Events (SSE) + +FastHTTP supports **Server-Sent Events (SSE)** — a one-way push protocol over plain HTTP where the server streams events to the client as they happen. + +SSE routes are defined with the `@app.sse()` decorator. The handler is called once per event and receives an [`SSEEvent`][sse-event] object. + +--- + +## Quick Start + +Connect to a public SSE stream and print each event: + +```python +from fasthttp import FastHTTP, SSEEvent + +app = FastHTTP() + + +@app.sse(url="https://stream.wikimedia.org/v2/stream/recentchange") +async def watch_wikipedia(event: SSEEvent) -> None: + import json + data = json.loads(event.data) + print(f"[{event.event}] {data['title']}") + + +if __name__ == "__main__": + app.run() +``` + +Run it: + +```bash +python main.py +``` + +Output: + +``` +INFO | fasthttp | FastHTTP started +INFO | fasthttp | Running 1 routes +INFO | fasthttp | SSE streams: 1 +INFO | fasthttp | SSE connecting: https://stream.wikimedia.org/v2/stream/recentchange +INFO | fasthttp | SSE connected: https://stream.wikimedia.org/v2/stream/recentchange +[message] Frank Sinatra +[message] Python (programming language) +[message] List of mountains by elevation +^C INFO | fasthttp | Interrupted by user +``` + +Press **Ctrl+C** to stop. + +--- + +## `@app.sse()` Signature + +```python +@app.sse( + url: str, + *, + headers: dict[str, str] | None = None, + reconnect: bool = False, + max_retries: int = 0, + tags: list[str] | None = None, +) +``` + +| Parameter | Type | Default | Description | +|-----------|------|---------|-------------| +| `url` | `str` | — | SSE endpoint URL (`http://` or `https://`). | +| `headers` | `dict[str, str]` | `None` | Extra headers sent with the connection request. | +| `reconnect` | `bool` | `False` | Automatically reconnect when the connection drops. | +| `max_retries` | `int` | `0` | Max reconnect attempts (ignored when `reconnect=False`). `-1` = unlimited. | +| `tags` | `list[str]` | `None` | Tags for grouping and filtering routes. | + +--- + +## The `SSEEvent` Object + +Every event carries these fields: + +| Field | Type | Description | +|-------|------|-------------| +| `event` | `str` | Event type from the server (`"message"` by default). | +| `data` | `str` | The event payload as a string. | +| `id` | `str \| None` | Event ID used for reconnection tracking (`Last-Event-ID`). | +| `retry` | `int \| None` | Reconnection time in milliseconds suggested by the server. | + +```python +@app.sse(url="https://api.example.com/events") +async def handle(event: SSEEvent) -> None: + print(f"Type: {event.event}") # "update", "alert", ... + print(f"Data: {event.data}") # the payload + print(f"ID: {event.id}") # for reconnection +``` + +--- + +## Custom Headers + +Pass `headers` for authentication or custom User-Agent: + +```python +@app.sse( + url="https://api.example.com/events", + headers={ + "Authorization": "Bearer your-token", + "User-Agent": "MyApp/1.0", + }, +) +async def handle(event: SSEEvent) -> None: + print(event.data) +``` + +--- + +## Auto-Reconnect + +When `reconnect=True`, FastHTTP reconnects with exponential backoff (`2^attempt` seconds, capped at 30s) if the connection drops. The `Last-Event-ID` header is sent automatically so the server can resume from where you left off. + +```python +@app.sse( + url="https://api.example.com/events", + reconnect=True, + max_retries=10, +) +async def handle(event: SSEEvent) -> None: + print(event.data) +``` + +Set `max_retries=-1` for unlimited reconnection attempts. + +--- + +## Running with HTTP Routes + +SSE streams run alongside HTTP requests and WebSocket connections. HTTP requests complete first, then the app stays alive until all SSE streams are done or interrupted. + +```python +from fasthttp import FastHTTP, SSEEvent +from fasthttp.response import Response + +app = FastHTTP() + + +@app.get(url="https://api.example.com/status") +async def health(resp: Response) -> dict: + return resp.json() + + +@app.sse(url="https://stream.wikimedia.org/v2/stream/recentchange") +async def watch(event: SSEEvent) -> None: + print(event.data) + + +if __name__ == "__main__": + app.run() # HTTP completes first, then SSE stays alive until Ctrl+C +``` + +--- + +## Local Development + +For testing without a remote server, create a local SSE server. See the [`examples/sse/local_server.py`](https://github.com/ndugram/fasthttp/blob/master/examples/sse/local_server.py) for a complete self-contained example. + +```bash +python examples/sse/local_server.py +``` + +--- + +## SSE vs WebSocket + +| Feature | SSE | WebSocket | +|---------|-----|-----------| +| Direction | Server → Client | Bidirectional | +| Protocol | Plain HTTP | WS / WSS | +| Message format | Text only (`data:` lines) | Text + Binary | +| Auto-reconnect | Built-in (browsers) / via `reconnect=True` | Manual | +| Use case | Live feeds, notifications, AI token streaming | Chat, gaming, real-time collaboration | + +Use SSE when you only need to **receive** events from a server. Use WebSocket when you also need to **send** data back. + +[sse-event]: ../../reference/fasthttp.md#fasthttp.sse.SSEEvent diff --git a/docs/ru/tutorial/index.md b/docs/ru/tutorial/index.md index ca12246..e2d591e 100644 --- a/docs/ru/tutorial/index.md +++ b/docs/ru/tutorial/index.md @@ -38,6 +38,8 @@ - [Middleware](middleware/index.md) - Глобальная логика запросов - [Безопасность](security/index.md) - Встроенная защита - [GraphQL](graphql/index.md) - Поддержка GraphQL +- [SSE](sse/index.md) - Server-Sent Events +- [WebSocket](websocket/index.md) - WebSocket соединения - [OpenAPI](openapi/index.md) - Swagger UI ## Как использовать diff --git a/docs/ru/tutorial/sse/index.md b/docs/ru/tutorial/sse/index.md new file mode 100644 index 0000000..77b37f7 --- /dev/null +++ b/docs/ru/tutorial/sse/index.md @@ -0,0 +1,183 @@ +# Server-Sent Events (SSE) + +FastHTTP поддерживает **Server-Sent Events (SSE)** — однонаправленный протокол «толкающих» уведомлений поверх обычного HTTP, где сервер стримит события по мере их возникновения. + +SSE-роуты определяются через декоратор `@app.sse()`. Хэндлер вызывается на каждое событие и получает объект [`SSEEvent`][sse-event]. + +--- + +## Быстрый старт + +Подключитесь к публичному SSE-потоку Wikimedia и печатайте каждое событие: + +```python +from fasthttp import FastHTTP, SSEEvent + +app = FastHTTP() + + +@app.sse(url="https://stream.wikimedia.org/v2/stream/recentchange") +async def watch_wikipedia(event: SSEEvent) -> None: + import json + data = json.loads(event.data) + print(f"[{event.event}] {data['title']}") + + +if __name__ == "__main__": + app.run() +``` + +Запустите: + +```bash +python main.py +``` + +Вывод: + +``` +INFO | fasthttp | FastHTTP started +INFO | fasthttp | Running 1 routes +INFO | fasthttp | SSE streams: 1 +INFO | fasthttp | SSE connecting: https://stream.wikimedia.org/v2/stream/recentchange +INFO | fasthttp | SSE connected: https://stream.wikimedia.org/v2/stream/recentchange +[message] Frank Sinatra +[message] Python (programming language) +[message] List of mountains by elevation +^C INFO | fasthttp | Interrupted by user +``` + +Нажмите **Ctrl+C** для остановки. + +--- + +## Сигнатура `@app.sse()` + +```python +@app.sse( + url: str, + *, + headers: dict[str, str] | None = None, + reconnect: bool = False, + max_retries: int = 0, + tags: list[str] | None = None, +) +``` + +| Параметр | Тип | По умолчанию | Описание | +|----------|-----|--------------|----------| +| `url` | `str` | — | URL SSE-эндпоинта (`http://` или `https://`). | +| `headers` | `dict[str, str]` | `None` | Дополнительные заголовки запроса. | +| `reconnect` | `bool` | `False` | Автопереподключение при разрыве. | +| `max_retries` | `int` | `0` | Максимум попыток переподключения. `-1` = безлимит. | +| `tags` | `list[str]` | `None` | Теги для группировки и фильтрации. | + +--- + +## Объект `SSEEvent` + +Каждое событие содержит: + +| Поле | Тип | Описание | +|------|-----|----------| +| `event` | `str` | Тип события (по умолчанию `"message"`). | +| `data` | `str` | Тело события в виде строки. | +| `id` | `str \| None` | ID события для отслеживания переподключения (`Last-Event-ID`). | +| `retry` | `int \| None` | Время переподключения в мс (от сервера). | + +```python +@app.sse(url="https://api.example.com/events") +async def handle(event: SSEEvent) -> None: + print(f"Тип: {event.event}") # "update", "alert", ... + print(f"Данные: {event.data}") # payload + print(f"ID: {event.id}") # для переподключения +``` + +--- + +## Пользовательские заголовки + +Передавайте `headers` для аутентификации или кастомного User-Agent: + +```python +@app.sse( + url="https://api.example.com/events", + headers={ + "Authorization": "Bearer your-token", + "User-Agent": "MyApp/1.0", + }, +) +async def handle(event: SSEEvent) -> None: + print(event.data) +``` + +--- + +## Автопереподключение + +При `reconnect=True` FastHTTP переподключается с экспоненциальной задержкой (`2^attempt` секунд, не более 30с). Заголовок `Last-Event-ID` отправляется автоматически — сервер может продолжить с последнего полученного события. + +```python +@app.sse( + url="https://api.example.com/events", + reconnect=True, + max_retries=10, +) +async def handle(event: SSEEvent) -> None: + print(event.data) +``` + +Установите `max_retries=-1` для безлимитных попыток. + +--- + +## Совместная работа с HTTP + +SSE-потоки выполняются вместе с HTTP-запросами и WebSocket-соединениями. HTTP завершается первым, затем приложение остаётся активным до завершения SSE-потоков или Ctrl+C. + +```python +from fasthttp import FastHTTP, SSEEvent +from fasthttp.response import Response + +app = FastHTTP() + + +@app.get(url="https://api.example.com/status") +async def health(resp: Response) -> dict: + return resp.json() + + +@app.sse(url="https://stream.wikimedia.org/v2/stream/recentchange") +async def watch(event: SSEEvent) -> None: + print(event.data) + + +if __name__ == "__main__": + app.run() # HTTP выполняется первым, SSE остаётся активным до Ctrl+C +``` + +--- + +## Локальная разработка + +Для тестирования без удалённого сервера запустите встроенный пример с локальным SSE-сервером: + +```bash +python examples/sse/local_server.py +``` + +--- + +## SSE vs WebSocket + +| Возможность | SSE | WebSocket | +|-------------|-----|-----------| +| Направление | Сервер → Клиент | Двустороннее | +| Протокол | Обычный HTTP | WS / WSS | +| Формат | Только текст (`data:` строки) | Текст + Бинарные | +| Автопереподключение | Встроено | Вручную | +| Сценарии | Ленты событий, уведомления, AI-токены | Чат, игры, совместная работа | + +Используйте SSE если нужно только **получать** события от сервера. Используйте WebSocket если нужно ещё и **отправлять** данные обратно. + +[sse-event]: ../../reference/fasthttp.md#fasthttp.sse.SSEEvent diff --git a/examples/README.md b/examples/README.md index 0a5f24a..90b47bd 100644 --- a/examples/README.md +++ b/examples/README.md @@ -13,6 +13,7 @@ All examples are tested and working. - **error** — automatic error handling and testing various error scenarios - **middleware** - test middleware for FastHTTP - **events** - event hooks for request/response lifecycle +- **sse** - Server-Sent Events (real-time event streams) - **pydantic** - pydantic response validation examples - **http2** - http2 test for fasthttp - **tags** - grouping and filtering requests by tags diff --git a/examples/sse/__init__.py b/examples/sse/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/examples/sse/basic.py b/examples/sse/basic.py new file mode 100644 index 0000000..3459a32 --- /dev/null +++ b/examples/sse/basic.py @@ -0,0 +1,29 @@ +""" +SSE client — Wikimedia recent changes stream. + +Connects to Wikimedia EventStreams (free, no auth) and prints +recent wiki edits as they happen. + +Usage: + python examples/sse/basic.py +""" +import json + +from fasthttp import FastHTTP, SSEEvent + +app = FastHTTP() + + +@app.sse(url="https://stream.wikimedia.org/v2/stream/recentchange") +async def watch_wikipedia(event: SSEEvent) -> None: + try: + data = json.loads(event.data) + title = data.get("title", "?") + user = data.get("user", "?") + print(f"[{event.event}] {title} — edited by {user}") + except json.JSONDecodeError: + print(f"[{event.event}] {event.data}") + + +if __name__ == "__main__": + app.run() diff --git a/examples/sse/reconnect.py b/examples/sse/reconnect.py new file mode 100644 index 0000000..6004bdc --- /dev/null +++ b/examples/sse/reconnect.py @@ -0,0 +1,29 @@ +""" +SSE with auto-reconnect and custom headers. + +Connects to an SSE endpoint with an auth token and automatically +reconnects up to 10 times if the connection drops. + +Usage: + python examples/sse/reconnect.py +""" +import json + +from fasthttp import FastHTTP, SSEEvent + +app = FastHTTP() + + +@app.sse( + url="https://stream.wikimedia.org/v2/stream/recentchange", + headers={"User-Agent": "FastHTTP-SSE/1.0"}, + reconnect=True, + max_retries=10, +) +async def watch_wikipedia(event: SSEEvent) -> None: + data = json.loads(event.data) + print(f"[{event.event}] {data.get('title', '?')}") + + +if __name__ == "__main__": + app.run() diff --git a/fasthttp/__init__.py b/fasthttp/__init__.py index f8b55d6..45c77ce 100644 --- a/fasthttp/__init__.py +++ b/fasthttp/__init__.py @@ -16,6 +16,7 @@ ) from .routing import Router from .session import AsyncSession +from .sse import SSEEvent from .websocket import WebSocket, WebSocketMessage __all__ = ( @@ -35,6 +36,7 @@ "OAuth2ClientCredentials", "RetryMiddleware", "Router", + "SSEEvent", "SessionMiddleware", "WebSocket", "WebSocketMessage", diff --git a/fasthttp/app.py b/fasthttp/app.py index 4882680..fff6293 100644 --- a/fasthttp/app.py +++ b/fasthttp/app.py @@ -45,6 +45,7 @@ from .openapi.urls import build_docs_urls from .routing import Route, Router from .security import Security +from .sse import SSEEvent from .websocket import WebSocket as FastHTTPWebSocket if TYPE_CHECKING: @@ -557,6 +558,7 @@ async def lifespan(app: FastHTTP): self.default_encoding = default_encoding self._ws_routes: list[dict] = [] + self._sse_routes: list[dict] = [] self.event_hooks = EventHooks() @@ -1235,6 +1237,95 @@ def decorator(func: Callable[..., object]) -> Callable[..., object]: return decorator + def sse( + self, + url: Annotated[ + str, + Doc( + """ + SSE endpoint URL. + """ + ), + ], + *, + headers: Annotated[ + dict[str, str] | None, + Doc( + """ + Additional headers sent with the SSE connection request. + """ + ), + ] = None, + reconnect: Annotated[ + bool, + Doc( + """ + Automatically reconnect when the connection drops. + + When enabled, the client will keep retrying until + ``max_retries`` is reached. + """ + ), + ] = False, + max_retries: Annotated[ + int, + Doc( + """ + Maximum number of reconnection attempts (ignored when + ``reconnect`` is False). + + Set to ``-1`` for unlimited retries. + """ + ), + ] = 0, + tags: Annotated[ + list[str] | None, + Doc( + """ + Tags for grouping and filtering requests. + """ + ), + ] = None, + ) -> Callable[[Callable[..., object]], Callable[..., object]]: + """ + Decorator for registering a Server-Sent Events (SSE) stream. + + The decorated async function receives an :class:`SSEEvent` for + every event emitted by the server. SSE streams run alongside + HTTP routes and WebSocket connections in ``app.run()``. + + Example: + .. code-block:: python + + from fasthttp import FastHTTP + from fasthttp.sse import SSEEvent + + app = FastHTTP() + + @app.sse(url="https://api.example.com/events") + async def handle_events(event: SSEEvent) -> None: + print(event.data) + """ + def decorator(func: Callable[..., object]) -> Callable[..., object]: + validate_handler(func=func) + resolved_url = url + if self.base_url and not url.startswith(("http://", "https://")): + resolved_url = self._resolve_url(url) + self._sse_routes.append( + { + "url": resolved_url, + "handler": func, + "headers": headers or {}, + "reconnect": reconnect, + "max_retries": max_retries, + "tags": tags or [], + } + ) + self.logger.debug("Registered SSE: %s", url) + return func + + return decorator + def on_request( self, func: Annotated[ @@ -1324,13 +1415,16 @@ async def _run(self, routes: list[Route] | None = None) -> None: # noqa: C901 routes = routes or self.routes http_total = len(routes) ws_total = len(self._ws_routes) - total = http_total + ws_total + sse_total = len(self._sse_routes) + total = http_total + ws_total + sse_total self.logger.info("Running %d routes", total) if http_total: self.logger.info("HTTP requests: %d", http_total) if ws_total: self.logger.info("WebSocket connections: %d", ws_total) + if sse_total: + self.logger.info("SSE streams: %d", sse_total) if self.http2_enabled: self.logger.info("HTTP/2 enabled") if self.proxy: @@ -1356,6 +1450,11 @@ async def _guarded(route: Route) -> tuple[Route, float, Response | None]: for ws_route in self._ws_routes ] + sse_futures = [ + asyncio.ensure_future(self._run_sse(sse_route)) + for sse_route in self._sse_routes + ] + if http_total > 0: if http_total > 1: if sys.version_info >= (3, 11): @@ -1383,16 +1482,129 @@ async def _guarded(route: Route) -> tuple[Route, float, Response | None]: self.logger.info("Done in %.2fs", total_time) - if ws_futures: + conn_futures = ws_futures + sse_futures + + if conn_futures: + labels = [] + if ws_futures: + labels.append("WebSocket connections") + if sse_futures: + labels.append("SSE streams") self.logger.info( - "WebSocket connections active. Press Ctrl+C to stop." + "%s active. Press Ctrl+C to stop.", + " and ".join(labels), ) - await asyncio.gather(*ws_futures, return_exceptions=True) + await asyncio.gather(*conn_futures, return_exceptions=True) async def _run_with_lifespan(self, routes: list[Route]) -> None: async with self.lifespan(self): # type: ignore await self._run(routes) + async def _run_sse(self, sse_route: dict) -> None: + url = sse_route["url"] + handler = sse_route["handler"] + request_headers = sse_route.get("headers", {}) + reconnect = sse_route.get("reconnect", False) + max_retries = sse_route.get("max_retries", 0) + + self.logger.info("SSE connecting: %s", url) + attempt = 0 + last_event_id: str | None = None + + while True: + try: + async with httpx.AsyncClient( + http2=self.http2_enabled, + proxy=self.proxy, + ) as client: + req_headers = dict(request_headers) + req_headers.setdefault("Accept", "text/event-stream") + req_headers.setdefault("Cache-Control", "no-cache") + if last_event_id: + req_headers["Last-Event-ID"] = last_event_id + + async with client.stream( + "GET", url, headers=req_headers + ) as resp: + resp.raise_for_status() + self.logger.info("SSE connected: %s", url) + + event_buf: list[str] = [] + event_type = "message" + event_id: str | None = None + + async for line in resp.aiter_lines(): + line = line.rstrip("\r\n") + + if line == "": + if event_buf: + event = SSEEvent( + event=event_type, + data="\n".join(event_buf), + id=event_id, + ) + if event.id: + last_event_id = event.id + await handler(event) + event_buf = [] + event_type = "message" + event_id = None + continue + + if line.startswith("data:"): + event_buf.append(line[5:].lstrip()) + elif line.startswith("event:"): + event_type = line[6:].strip() + elif line.startswith("id:"): + val = line[3:].strip() + if val and not val.startswith("\0"): + event_id = val + elif line.startswith(":"): + continue + + self.logger.info("SSE stream ended: %s", url) + break + + except httpx.HTTPStatusError as e: + self.logger.error( + "SSE HTTP error for %s: %d %s", + url, + e.response.status_code, + e.response.text, + ) + if reconnect and (max_retries == -1 or attempt < max_retries): + attempt += 1 + wait = min(2**attempt, 30) + self.logger.warning( + "SSE reconnecting in %ds (attempt %d)...", + wait, + attempt, + ) + await asyncio.sleep(wait) + continue + break + except httpx.RequestError as e: + self.logger.error("SSE request error for %s: %s", url, e) + if reconnect and (max_retries == -1 or attempt < max_retries): + attempt += 1 + wait = min(2**attempt, 30) + self.logger.warning( + "SSE reconnecting in %ds (attempt %d)...", + wait, + attempt, + ) + await asyncio.sleep(wait) + continue + break + except Exception as e: + self.logger.error( + "SSE error for %s: %s %s", + url, + type(e).__name__, + e, + ) + break + async def _run_ws(self, ws_route: dict) -> None: """Connect and run a single WebSocket handler.""" diff --git a/fasthttp/sse.py b/fasthttp/sse.py new file mode 100644 index 0000000..da8c679 --- /dev/null +++ b/fasthttp/sse.py @@ -0,0 +1,12 @@ +from __future__ import annotations + +from dataclasses import dataclass, field + + +@dataclass +class SSEEvent: + event: str = "message" + data: str = "" + id: str | None = None + retry: int | None = None + _raw: list[str] = field(default_factory=list, repr=False) diff --git a/mkdocs.yml b/mkdocs.yml index b30fa34..60607e2 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -119,6 +119,8 @@ nav: - en/tutorial/graphql/index.md - en/tutorial/graphql/queries.md - en/tutorial/graphql/mutations.md + - SSE: + - en/tutorial/sse/index.md - WebSocket: - en/tutorial/websocket/index.md - OpenAPI: @@ -185,6 +187,8 @@ nav: - ru/tutorial/graphql/index.md - ru/tutorial/graphql/queries.md - ru/tutorial/graphql/mutations.md + - SSE: + - ru/tutorial/sse/index.md - WebSocket: - ru/tutorial/websocket/index.md - OpenAPI: