Type two characters. Titles and headings are searched, not body text.

Chat room

The live-page pattern as a project: features, SQLite, tests.

Chat room is a project, not a file: two features, SQLite, a shared identity helper, and tests. It is too big to type along, so this tutorial walks the source. Clone it, run it, then read it top to bottom with this page beside you.

It assumes you have done Realtime tiles. The three jobs are the same: a page, an open stream, short commands. What changes is how the code is arranged once there is more than one page.

1. Run it#

git clone https://github.com/stariodev/stario.git
cd stario/examples/chat-room
uv sync
uv run --with watchfiles watchfiles 'python -m app.main' app/

Open http://127.0.0.1:8000 in two windows. Create a room in one; it appears in the other. Open the room in both and send a message. Run the tests with uv run pytest.

2. The tree#

app/
  main.py                 bootstrap: config, database, relay, features
  config.py               Config.from_env()
  db.py                   Database: one SQLite connection
  assets.py               ASSETS = Assets(...)
  common/
    identity.py           who is this tab: user_id, username, colour
    shell.py              the document shell every page uses
  features/
    lobby/
      urls.py             LOBBY, LIVE
      signals.py          the lobby's signal names
      subjects.py         relay subject names
      views.py            lobby_view, lobby_live_view
      handlers.py         LobbyPage: register, show, live, create_room, delete_room
    room/
      urls.py             ROOM, LIVE, SEND, TYPING, CREATE, DELETE
      signals.py          the room's signal names
      subjects.py         relay subject names
      models.py           Room, Message, User
      data.py             SCHEMA and the queries
      views.py            room_view, room_live_view
      handlers.py         RoomPage: register, show, live, send_message, typing
tests/
  conftest.py             client fixture: TestClient(bootstrap)
  test_assets.py
  test_lobby.py
  test_room.py

One folder per page. Every feature has urls.py, views.py, and handlers.py; signals and subjects get their own files when the page uses them; a feature that owns tables also has models.py and data.py. The lobby owns no tables; it reads the room queries.

3. Bootstrap#

app/main.py is the only file that knows about everything. It reads config, opens the database, applies the schema, creates the Relay, attaches static files, and registers each feature.

async def bootstrap(app: App, span: Span):
    config = Config.from_env()
 
    db = Database(config.db_path)
    db.apply_schema(room_data.SCHEMA)
 
    relay: Relay[dict] = Relay()
 
    with span.step("static_assets") as s:
        s.attrs(await ASSETS.attach(app))
 
    LobbyPage(db, relay).register(app)
    RoomPage(db, relay).register(app)
    yield

In tiles the page was one class holding game and relay. Here it is the same idea, one level up: each feature’s handlers.py holds one page class — its fields are the shared objects, and register adds every route.

@dataclass(frozen=True, slots=True)
class RoomPage:
    db: Database
    relay: Relay[dict]
 
    def register(self, app: App) -> None:
        app.add(ROOM, self.show)
        app.add(LIVE, self.live)
        app.add(SEND, self.send_message)
        app.add(TYPING, self.typing)

Nothing under app/features/ imports bootstrap. The import graph points one way: main.py imports the features, the features import common, and urls.py imports only stario.

4. Routes#

Routes live in urls.py so views and handlers can both import them without importing each other.

# app/features/room/urls.py
ROOMS = "/rooms"
ROOM_PATH = ROOMS + "/{room_id}"
ROOM = Route("GET", ROOM_PATH)
LIVE = Route("GET", ROOM_PATH + "/live")
SEND = Route("POST", ROOM_PATH + "/send")
TYPING = Route("POST", ROOM_PATH + "/typing")
CREATE = Route("POST", ROOMS)
DELETE = Route("DELETE", ROOM_PATH)

Each page has the same three kinds of route. GET /rooms/{room_id} is the page. GET /rooms/{room_id}/live is the stream. The POST and DELETE routes are commands. The lobby is the same: GET /, GET /live, and the create and delete commands, which it imports from the room feature because they change rooms.

5. The stream#

The live method in app/features/room/handlers.py is tiles’ stream handler with a database behind it. Read it from the top with this in mind:

  1. Load the room. If it is gone, SSE(w).navigate(LOBBY.href()) and return.

  2. Read the tab’s identity from the signals. Without one, SSE(w).navigate(ROOM.href(...)) reloads the room page, where show mints it.

  3. Subscribe to subjects.room_events(room.id), the wildcard that matches presence, message, typing, and deleted. Only then add the user to the room and publish on subjects.presence(room.id), so this tab’s own join is not missed.

  4. Send the first patch.

  5. On every relay event, check that the room still exists. If it was deleted, sse.navigate(LOBBY.href()) and break. Otherwise patch again.

  6. After the loop ends, remove the user and publish on subjects.presence(room.id) again, so the other tabs repaint the roster.

The patch is the same each time, and it is narrower than a full document:

def patch_room(sse, *, room, user_id, username, color, db):
    sse.patch_elements(
        room_live_view(
            room,
            user_id,
            messages=data.list_messages(db, room.id),
            users=data.list_users(db, room.id),
        ),
        selector=ROOM_LIVE_SELECTOR,  # "#room-live"
        mode="outer",
    )

room_view renders the whole document once; room_live_view renders the #room-live region inside it — the messages, the user list, the typing line. The stream patches that region, and Datastar morphs the difference. Live pages starts from the whole-document version; a selector is the same idea with a smaller patch.

6. The commands#

send_message validates, replies, then works, in that order:

async def send_message(self, c: Context, w: Writer) -> None:
    room = room_from_route(c, self.db)
    if room is None:
        SSE(w).navigate(LOBBY.href())
        return
 
    signals = await read_chat_signals(c)
    if not signals.user_id or not data.user_exists(self.db, signals.user_id, room.id):
        SSE(w).navigate(ROOM.href(room_id=room.id))
        return
 
    text = signals.message.strip()[:MAX_MESSAGE_LENGTH]
    if not text:
        w.empty()
        return
 
    w.empty()
    data.add_message(self.db, Message(...))
    self.relay.publish(subjects.message(room.id), {})

A missing room sends the tab to the lobby with SSE(w).navigate; a stranger reloads the room page, where show mints an identity. The answer is navigate rather than w.redirect because Datastar’s fetch follows a 303 without changing the address bar — Live pages shows the difference. A valid command replies 204 before it writes to the database. The new message reaches every tab through the stream, including the sender’s.

delete_room shows the one twist. After it deletes a room, it publishes subjects.deleted(room_id). Every live loop for that room wakes on the wildcard, finds the room gone, and calls sse.navigate(LOBBY.href()). Open tabs move to the lobby without the command knowing who they were.

7. Identity#

There is no login. app/common/identity.py mints a user_id, a name, and a colour on the first page view and stores them as Datastar signals. A small data.init script keeps them in sessionStorage, so a reload in the same tab keeps the same identity and two tabs get two. Every request from the page carries the signals, so read_signals(c.req) is how a handler learns who is asking.

This is a demo identity. Sessions and login shows the signed-cookie version.

8. Tests#

tests/conftest.py is a short fixture that opens TestClient(bootstrap). The tests run the real bootstrap, so they use the same routes, schema, and relay as production, with CHAT_DB_PATH left at its :memory: default.

The stream tests open a stream and read a patch — test_lobby_live in tests/test_lobby.py does it for the lobby. A send-repaints test in the same shape looks like this:

async def test_room_send_repaints_the_stream(client):
    await client.post("/rooms", json={"room_title": "General"})
    signals = json.dumps({"user_id": "u1", "username": "HappyFox", "color": "#e74c3c"})
    async with client.stream(
        "GET",
        "/rooms/general/live",
        params={"datastar": signals},
        timeout=2,
    ) as live:
        events = live.iter_events()
        first = await anext(events)
        assert "General" in first["data"]
        await anext(events)  # the repaint for this tab's own join
 
        r = await client.post(
            "/rooms/general/send",
            json={"user_id": "u1", "username": "HappyFox", "color": "#e74c3c", "message": "hello room"},
        )
        assert r.status_code == 204
 
        repaint = await anext(events)
        assert "hello room" in repaint["data"]

Open the stream, read the first render, send a command, read the repaint. One test covers the page, the command, the relay, and the render. Testing with TestClient explains stream and iter_events.

By the way#

  • Config.from_env() reads CHAT_DB_PATH and nothing else. Stario does not load dotenv files; set variables in the shell or the process manager.

  • subjects.py holds the relay subject names — LOBBY_CHANGED is a constant, and the room’s are functions such as subjects.presence(room_id) — so a typo is a NameError and not a silent miss. A publish carries {} when every subscriber should just repaint.

  • The lobby’s stream subscribes to room.* and lobby.*. A message in any room changes the online count, so the lobby repaints too.

Next: Structuring apps states the layout rules this example follows.

Stario

Contents

Theme

Made by Adam with Stario