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 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 or from the body on POST, PUT, and other methods (read_signals).

Read signals as JSON (`read_signals`)

python

from stario import Context, Writer, datastar as ds

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

    signals = await ds.read_signals(c.req)

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

    ds.sse.patch_signals(w, {"count": count + 1})  # SSE response must already be open in a real route

read_signals is a thin json.loads—fine for prototypes; treat the result as untrusted until you validate.

Form validation with signals

Stario does not validate the payload—you map failures to HttpException, patch_signals, or plain responses. You can use await ds.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.

Read the wire the same way on every method: datastar query on GET, await req.body() on POST, PUT, and other methods (Request).
Decode and validate in one place—usually the typed helpers below (TypeAdapter.validate_json, msgspec, cattrs, …).
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 or await req.body()) 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(...).

python

from pydantic import BaseModel, TypeAdapter, ValidationError

from stario import HttpException, Request

class MySignals(BaseModel):

    count: int

_signals = TypeAdapter(MySignals)

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

    if req.method == "GET":

        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

msgspec

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

python

import msgspec

from stario import HttpException, Request

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

    if req.method == "GET":

        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

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 ds.read_signals(req) and structure the dict; if you json.loads manually, treat "" / b"" as {} before decoding.

python

import cattrs

from attrs import define

from cattrs.errors import BaseValidationError

from stario import HttpException, Request, datastar as ds

@define

class PageSignals:

    count: int

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

    signals = await ds.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.

Tests: TestClient can post files= and data= together for multipart/form-data; see Testing with TestClient.

Write: `patch_signals`

ds.sse.patch_signals merges JSON into the client signal store. It accepts a dict (JSON object) or bytes of a JSON object (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_json(obj) returns bytes for patch_signals.

python

from pydantic import BaseModel, TypeAdapter

from stario import Writer, datastar as ds

class MySignals(BaseModel):

    count: int

_signals = TypeAdapter(MySignals)

def patch_signals_pydantic(w: Writer, obj: MySignals) -> None:

    ds.sse.patch_signals(w, _signals.dump_json(obj))

msgspec

python

import msgspec

from msgspec import Struct

from stario import Writer, datastar as ds

def patch_signals_msgspec(w: Writer, obj: Struct) -> None:

    ds.sse.patch_signals(w, msgspec.json.encode(obj))

attrs

python

from attrs import asdict

from stario import Writer, datastar as ds

def patch_signals_attrs(w: Writer, obj: object) -> None:

    ds.sse.patch_signals(w, asdict(obj))

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