Datastar from Python

Datastar is a small browser library. It reads data-* attributes on your HTML, sends requests when they say so, and morphs the HTML the server returns into the page. There is no client build step and no JavaScript of your own. stario.datastar is the Python half: it writes the attributes, reads the signals a request carries, and streams the patches back.

This page covers the five things you do with it. The Datastar reference lists every helper.

from stario.datastar import SSE, ModuleScript, at, data, read_signals

1. Load the client#

from stario.datastar import ModuleScript

from stario import h

h.Head(ModuleScript())

ModuleScript() points at a pinned CDN build, which is fine while you experiment. In production, self-host the file and pass src=, or serve it through Assets as the tutorials do.

2. Attributes and actions in views#

data.* returns an attribute. at.* returns the action string that goes inside one. The URL comes from the route, so a renamed path changes in one place.

from stario import Route

from stario.datastar import at, data

from stario import h

SUBSCRIBE = Route("GET /subscribe")

PAINT = Route("POST /click")

def board():

    return h.Div(

        data.signals({"user_id": ""}, if_missing=True),

        data.init(at.get(SUBSCRIBE.href(), retry="always")),

        h.Button(data.on("click", at.post(PAINT.href())), "Paint"),

    )

The browser sees data-signals, data-init, data-on:click, @get, and @post. Two more you will use often: data.bind("email") ties an input to a signal, and data.text("$count") shows one.

3. Read what the page sent#

Every Datastar request carries the page's signals as one JSON object. read_signals(c.req) returns it as a dict. Treat it like any other input from a browser: it is whatever the page had, including anything the user changed in the developer tools.

from stario.datastar import SSE, read_signals

async def increment(c, w):

    signals = await read_signals(c.req)

    count = int(signals.get("count") or 0)

    SSE(w).patch_signals({"count": count + 1})

Signal names are snake_case on the wire, and nested state is a nested object, not a dotted key. Malformed JSON raises StarioError, which is a 500, because a real browser does not send it; for a value that is merely wrong, check it and answer w.text("...", status=400).

4. Validate with the library you already use#

Stario does not bundle a validator. Read the same bytes read_signals reads and hand them to pydantic, msgspec, or attrs. GET and DELETE carry the signals in the datastar query parameter; other methods carry them in the body.

from pydantic import BaseModel, TypeAdapter, ValidationError

from stario.http import Request

class Counter(BaseModel):

    count: int

_counter = TypeAdapter(Counter)

async def read_counter(req: Request) -> Counter:

    raw = req.query.get("datastar") or "" if req.method in ("GET", "DELETE") else await req.body()

    try:

        return _counter.validate_json(raw)

    except ValidationError as exc:

        raise SignalsInvalid("Invalid signals") from exc

SignalsInvalid is a small exception of yours, mapped to a 400 with catch_errors — Error handling shows the pattern.

Keep one TypeAdapter per shape at module level. An empty body fails validation and becomes a 400, where read_signals would have returned {}. With msgspec, call msgspec.json.decode(raw, type=Counter) on the same raw. With attrs and cattrs, start from await read_signals(req) and structure the dict.

5. Patch the page over SSE#

SSE(w) turns the response into a stream of events. patch_elements sends HTML that Datastar morphs into the page by element id. patch_signals merges a JSON object into the page's signals. A subscribe handler calls them many times over minutes; a command usually calls none and replies 204, because the open stream will repaint.

async def subscribe(c, w):

    sse = SSE(w)

    async with relay.subscribe("board") as live:

        sse.patch_elements(board_view(game))

        async for _ in c.alive(live):

            sse.patch_elements(board_view(game))

To send a typed object back, dump it first: sse.patch_signals(_counter.dump_python(obj)), msgspec.to_builtins(obj), or attrs.asdict(obj).

Once SSE(w) has written, the response is a stream; do not call a w.* response method on the same writer. A response finishes once.

Files#

A file input bound with data.bind arrives as a signal, not as multipart/form-data: an object with name, base64 contents, and an optional mime. FileSignal in the Datastar reference is its type. Cap the size and check the type before you decode; User uploads covers storing the bytes.

By the way#

• One open stream plus short commands is the live-page pattern. Realtime tiles builds it in one file.

• A command that should end on another page finishes with SSE(w).navigate(url). Do not answer a Datastar request with a 303; fetch follows it and the page morphs instead of moving. Live pages explains why.

• Attribute modifiers, the Pro helpers, and the client runtime are documented at data-star.dev.

The complete example is examples/tiles/main.py.