Real-Time Web

Overview

Lexigram supports two real-time primitives: SSE (one-way server→client) and WebSocket (bidirectional). Both are class-based handlers registered by decorators. HTMX can consume SSE streams for declarative live updates.

SSE with HTMX

Subclass AbstractSSEHandler, decorate the class with @sse_endpoint(path), and override stream(request) as an async generator yielding event dicts:

from lexigram.web.sse import sse_endpoint
from lexigram.web.sse.handler import AbstractSSEHandler
from starlette.requests import Request

@sse_endpoint("/events/{channel}", heartbeat_interval=15)
class LiveFeedHandler(AbstractSSEHandler):
    event_types = ["message"]

    async def stream(self, request: Request):
        channel = request.path_params["channel"]
        async for update in subscribe(channel):        # your pub/sub source
            yield {"event": "message", "data": update}

There is no push() — each client runs its own generator. To fan out to many clients, read from a shared async queue or pub/sub channel inside stream() (the built-in broadcast() raises NotImplementedError). Override on_connect/on_disconnect for lifecycle; max_connections caps concurrency.

HTMX client:

<div hx-sse="connect:/events/dashboard swap:message">
  <div hx-get="/partials/dashboard" hx-trigger="load">Loading…</div>
</div>

WebSocket Handler

@websocket_handler(path) decorates the handler class (not methods). Subclass AbstractWebSocketHandler and implement on_connect / on_message / on_disconnect; on_message receives an already-parsed JSON dict:

from lexigram.web.websocket.decorators import websocket_handler
from lexigram.web.websocket.handler import AbstractWebSocketHandler

@websocket_handler("/ws/chat/{room_id}")
class ChatHandler(AbstractWebSocketHandler):
    async def on_connect(self, websocket):
        await websocket.accept()                       # required

    async def on_message(self, websocket, message: dict):
        await self.broadcast(message)                  # or send per-connection

    async def on_disconnect(self, websocket):
        pass

For raw receive loops use while True: message = await websocket.receive_text() — there is no iter_text() async iterator on the connection.

Common Mistakes

• Decorating methods instead of classes — both @sse_endpoint and @websocket_handler wrap the class
• Calling push(...) or broadcast(...) on SSE handlers — SSE is unidirectional; share a queue/pub-sub in stream()
• Forgetting await websocket.accept() in WebSocket on_connect — connection never opens
• Blocking the SSE generator with sync I/O — keep it async throughout