Datastar: Reading and writing signals

Stario does not bundle Pydantic, msgspec, attrs/cattrs, or any other validation or JSON stack—the framework core stays minimal. Why we do that (and what we lean on instead): No validation layer in the framework.

This page is the single Datastar recipe: read_signals, validation (including form-style flows), file signals, typed reads with libraries you add, and SSE updates via SSE(w).patch_signals.

On the wire, Datastar keeps page signals as one JSON object per action (data-signals on the client); the server reads that blob from the datastar query on GET and DELETE, or from the body on POST, PUT, PATCH, and other methods that send a body (Datastar #1146, read_signals).

Import helpers from stario.datastar (SSE, at, data, read_signals, …).

from stario.datastar import SSE, read_signals

Read signals as JSON (read_signals)

from stario import Context, Writer

from stario.datastar import SSE, read_signals

async def bump(c: Context, w: Writer) -> None:

    signals = await read_signals(c.req)

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

    sse = SSE(w)

    sse.patch_signals({"count": count + 1})

read_signals is a thin json.loads—fine for prototypes; treat the result as untrusted until you validate. Malformed JSON or a non-object top-level value raises StarioError from read_signals (typically a 500 unless you register on_error). For client mistakes, validate in your helper and raise HttpException(400) as in the typed examples below.

Signal names on the wire use Python snake_case. Nested state belongs in nested JSON objects, not dotted top-level keys.

Form validation with signals

Stario does not validate the payload—you map failures to HttpException, patch_signals, or plain responses. You can use await read_signals(req) and then validate the dict, or read raw query/body bytes and decode in one step with Pydantic, msgspec, or cattrs—both are valid; typed decoders often prefer the raw path.

1. Read the wire the same way on every method: datastar query on GET and DELETE, await req.body() on POST, PUT, PATCH, and other methods that carry signals in the body (Request).

2. Decode and validate in one place—usually the typed helpers below (TypeAdapter.validate_json, msgspec, cattrs, …).

3. On success, run domain logic and respond or patch. On validation failure, return 400 or patch_signals with field-level detail so the UI updates without a full navigation.

Why the framework stays out of validation: No validation layer in the framework.

Typed signals (Pydantic, msgspec, cattrs)

Read the same string or bytes Stario exposes (query datastar on GET/DELETE, or await req.body() on other methods) and decode with your library. Empty or malformed payloads fail inside the validator—you map that to HttpException as you like.

Pydantic

Use TypeAdapter with validate_json on the wire. One adapter per shape at module scope (or a registry when you have many)—not a new TypeAdapter every request.

That covers BaseModel, TypedDict, @dataclass, unions, and anything else TypeAdapter supports. validate_json accepts str or bytes; configure aliases and field names on the schema you pass to TypeAdapter(...).

from pydantic import BaseModel, TypeAdapter, ValidationError

from stario import HttpException

from stario.http import Request

class MySignals(BaseModel):

    count: int

_signals = TypeAdapter(MySignals)

async def read_signals_pydantic(req: Request) -> MySignals:

    if req.method.upper() in ("GET", "DELETE"):

        raw: str | bytes = req.query.get("datastar") or ""

    else:

        raw = await req.body()

    try:

        return _signals.validate_json(raw)

    except ValidationError as exc:

        raise HttpException(400, "Invalid signals") from exc

Unlike read_signals, an empty query or body fails validation here and maps to 400, not {}.

msgspec

msgspec.json.decode accepts str or bytes-like buffers—you can pass the datastar query string as-is on GET and DELETE, or await req.body() on methods that send signals in the body, without an extra .encode("utf-8") unless you prefer bytes everywhere.

import msgspec

from stario import HttpException

from stario.http import Request

async def signals_msgspec[T](req: Request, typ: type[T]) -> T:

    if req.method.upper() in ("GET", "DELETE"):

        raw: str | bytes = req.query.get("datastar") or ""

    else:

        raw = await req.body()

    try:

        return msgspec.json.decode(raw, type=typ)

    except msgspec.DecodeError as exc:

        raise HttpException(400, "Invalid signals") from exc

Unlike read_signals, an empty query or body fails validation here and maps to 400, not {}.

attrs and cattrs

cattrs does not parse JSON text by itself—it structures already-parsed Python values (usually a dict from json.loads) into attrs types. To match read_signals empty-query/body behavior ({}), start from await read_signals(req) and structure the dict; if you json.loads manually, treat "" / b"" as {} before decoding.

import cattrs

from attrs import define

from cattrs.errors import BaseValidationError

from stario import HttpException

from stario.http import Request

from stario.datastar import read_signals

@define

class PageSignals:

    count: int

async def signals_attrs(req: Request) -> PageSignals:

    signals = await read_signals(req)

    try:

        return cattrs.structure(signals, PageSignals)

    except (BaseValidationError, TypeError) as exc:

        raise HttpException(400, "Invalid signals") from exc

File uploads

For Datastar-driven pages, file inputs are usually file signals: a structured object with a name field (filename on the wire), base64 contents, and optional mime—see FileSignal in Datastar—not raw multipart/form-data unless you post a classic HTML form.

• Stario — Datastar describes FileSignal and helpers that match Datastar’s model.

• Upstream — Datastar’s data-bind docs cover how uploads land in signals (File Uploads in the Datastar reference).

Treat that JSON as untrusted: cap size, allowed types, and shape in your code—same as any other signal. Use await req.body() or read_signals, then validate with Pydantic or another library you add.

For classic multipart HTML forms, Stario exposes the raw Request only—no built-in multipart parser. Buffer or stream once, then parse with an explicit library and limits you control.

Once you have bytes to persist or serve from disk or object storage (naming, authorization, download responses), see User uploads and storage.

Tests: TestClient can post files= and data= together for multipart/form-data; for signal-encoded bodies use GET/DELETE params={"datastar": '{"count": 1}'} or POST content=b'{"count": 1}' — see Testing with TestClient.

Write: SSE(w).patch_signals

Create one SSE per response. patch_signals merges a mapping into the client signal store—top-level keys must be a JSON object (dict / Mapping), not raw bytes (SSE.patch_signals).

Keep helpers generic—one pattern per library, not one function per concrete type.

Pydantic

Use the same TypeAdapter you built for reads—dump_python(obj) (or model_dump()) for patch_signals.

from pydantic import BaseModel, TypeAdapter

from stario.datastar import SSE

class MySignals(BaseModel):

    count: int

_signals = TypeAdapter(MySignals)

def patch_signals_pydantic(sse: SSE, obj: MySignals) -> None:

    sse.patch_signals(_signals.dump_python(obj))

msgspec

import msgspec

from msgspec import Struct

from stario.datastar import SSE

def patch_signals_msgspec(sse: SSE, obj: Struct) -> None:

    sse.patch_signals(msgspec.to_builtins(obj))

attrs

from attrs import asdict

from stario.datastar import SSE

def patch_signals_attrs(sse: SSE, obj: object) -> None:

    sse.patch_signals(asdict(obj))

Long-lived streams

For SSE routes that stay open, pair SSE(w) with c.alive() so disconnect and server shutdown end the loop cleanly:

async with c.alive():

    sse = SSE(w)

    sse.patch_signals({"status": "connected"})

    # …

Or stream from a source until the client leaves: async for item in c.alive(live): … (see the chat-room example).

Explanation: No validation layer in the framework · Reference: Datastar · Routing