How I want to use Stardust in a web app

shopping.py

"""A tiny shopping-list web app, written the way I would expect to write Stardust code."""

import stardust
from stario import App, Context, Route, Span, Writer, responses
from stario.datastar import SSE, at, data, read_signals
from stario.datastar.format import string_literal
from stario.markup import html as h

NEEDED = {
    "find": ["?name", "?need"],
    "where": [["?i", "item/name", "?name"], ["?i", "item/need", "?need"], [">", "?need", 0]],
    "orderBy": ["?name"],
}

# Ensure a user only buys an item if we need at least one.
BUY_ONE = {
    "query": {
        "find": ["?i", "?left"],
        "where": [
            ["?i", "item/name", "?name"],
            ["?i", "item/need", "?need"],
            [">", "?need", 0],
            [["-", "?need", 1], "?left"],
        ],
        "parameters": {"?name": ""},
    },
    "patch": {"?i": {"item/need": "?left"}},
}


async def set_up(db: stardust.Database) -> None:
    await db.define.query("needed", NEEDED)
    await db.define.mutation("buy_one", BUY_ONE)
    await db.patch({"#_milk": {"item/name": "Milk", "item/need": 2}})


async def register(app: App, db: stardust.Database) -> None:
    async def show(c: Context, w: Writer) -> None:
        # An ordinary request: query and render.
        async with db.query("needed") as query, query.run() as rows:
            responses.html(w, page(await rows.all_pages()))

    async def buy(c: Context, w: Writer) -> None:
        name = (await read_signals(c.req))["name"]
        try:
            async with db.mutate("buy_one", name=name) as mutate:
                await mutate.run()
        except stardust.NoMatch:
            # Nothing matched: the item was just bought, or isn't on the list (any more).
            SSE(w).patch_elements(h.Div({"id": "flash"}, f"Sorry, {name} isn't available."))
            return
        responses.empty(w)  # success: the live connection shows the change to everyone

    async def live(c: Context, w: Writer) -> None:
        # The one long-lived endpoint: the list now, and again after every change anyone makes.
        sse = SSE(w)
        async with db.query("needed") as query, query.subscribe() as subscription:
            async for event in c.alive(subscription.events()):
                sse.patch_elements(list_view(event.result))

    app.add(HOME, show)
    app.add(BUY, buy)
    app.add(LIVE, live)


async def bootstrap(app: App, span: Span):
    async with stardust.open("list.db") as db:
        await set_up(db)
        await register(app, db)
        yield


# STARIO ONLY from here on
# ============================================================================================

HOME = Route("GET /")
LIVE = Route("GET /live")
BUY = Route("POST /buy")


def list_view(rows):
    return h.Ul(
        {"id": "list"},
        *[
            h.Li(
                f"{row.name} x{row.need} ",
                h.Button(
                    data.on(
                        "click", at.post(BUY.href(), payload={"name": string_literal(row.name)})
                    ),
                    "Buy one",
                ),
            )
            for row in rows
        ],
    )


def page(rows):
    return h.HtmlDocument(
        h.Head(h.Title("Shopping list")),
        h.Body(
            {"id": "screen"},
            data.init(at.get(LIVE.href(), retry="always")),  # open the live connection
            h.Div({"id": "flash"}),
            list_view(rows),
        ),
    )