SSE

Live feed over SSE

This example streams HTML to the browser after the page loads — a feed of notifications that appear one by one, with no client-side JavaScript. You return anEventStreamfrom a route, and an async generator yields events over a long-lived connection. Each yielded value can be a plain string, a structured SSEEvent (custom event name and id), or a rendered Fragment(...)that htmx swaps into the page.

Minimal example

Two routes: one renders the page shell, the other returns anEventStreamwhose generator yields a mix ofSSEEvent and Fragmentvalues.

import asyncio
from pathlib import Path

from chirp import App, AppConfig, EventStream, Fragment, SSEEvent, Template

config = AppConfig(
    template_dir=Path(__file__).parent / "templates",
    worker_mode="async",
    sse_close_event="close",
)
app = App(config=config)

_NOTIFICATIONS = [
    {"title": "Welcome", "message": "You are now connected to the live feed."},
    {"title": "Update", "message": "New deployment started."},
    {"title": "Alert", "message": "CPU usage above 90% on worker-3."},
    {"title": "Resolved", "message": "CPU usage back to normal."},
]


@app.route("/")
def index():
    return Template("feed.html")


@app.route("/events", referenced=True)
def events():
    async def generate():
        # Structured SSEEvent — custom event name, kept off the htmx message channel.
        yield SSEEvent(data="connected", event="status")

        # Fragment events — rendered HTML pushed to the browser and swapped by htmx.
        for notification in _NOTIFICATIONS:
            await asyncio.sleep(1.5)
            yield Fragment(
                "feed.html",
                "notification",
                title=notification["title"],
                message=notification["message"],
            )

    return EventStream(generate())


if __name__ == "__main__":
    app.run()

Source: examples/standalone/sse/app.py.

Setworker_mode="async"— the stream holds a connection open and awaits between events. The/events route is marked referenced=Truebecause htmx reaches it through ansse-connectattribute, not a link the contract checker can see, so the flag keeps it out of the orphan-route report. AFragment's target becomes the SSE event name, so ansse-swap="notification"element receives only those updates.

Run it

1
Start the app
```
PYTHONPATH=src python examples/standalone/sse/app.py
```
Open http://127.0.0.1:8000/and watch notifications stream in.
2
Run the test
```
pytest examples/standalone/sse/
```

Live feed over SSE

Minimal example

Run it

Start the app

Run the test