Skip to content

SyncModel

SyncModel

SyncModel(*args: Any, **kwargs: Any)

Bases: BaseModel, Reactive[SyncModelCtx]

The reactive counterpart of pydantic.BaseModel.

Subclass it instead of BaseModel, everything from Pydantic still works. When live, assigning a field is validated and written to the JSON file, and changes to the file are applied in place. Created directly, an instance is inert and acts like a regular Pydantic model.

Example:

from syncwave import SyncList, SyncModel, Syncwave

syncwave = Syncwave()


class Settings(SyncModel):
    volume: int = 8
    recent_files: SyncList[str] = SyncList()


settings = syncwave.create_store(Settings, name="settings")
settings.volume = 7  # written to settings.json
settings.recent_files.append("notes.txt")  # written too
print(settings.recent_files)  # ['notes.txt']


Usage Documentation

The SyncModel Class

Source code in src/syncwave/reactive.py
109
110
111
112
def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Raise `TypeError`: `Reactive` itself cannot be instantiated."""
    if type(self) is Reactive:
        raise TypeError("`Reactive` is a base class and cannot be instantiated.")

SyncRoot

SyncRoot(*args: Any, **kwargs: Any)

Bases: SyncModel, RootModel, Generic[RootModelRootType]

The reactive counterpart of pydantic.RootModel.

A SyncModel holding a single value in its root field. The JSON file holds the bare value.

Example:

from syncwave import SyncRoot, Syncwave

syncwave = Syncwave()


class Locale(SyncRoot[str]): ...


locale = syncwave.create_store(Locale, name="locale")
locale.root = "en-US"  # written to locale.json as "en-US"


Usage Documentation

Single Values with SyncRoot

Source code in src/syncwave/reactive.py
109
110
111
112
def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Raise `TypeError`: `Reactive` itself cannot be instantiated."""
    if type(self) is Reactive:
        raise TypeError("`Reactive` is a base class and cannot be instantiated.")

sync_dataclass

sync_dataclass(
    *,
    init: Literal[False] = False,
    repr: bool = True,
    eq: bool = True,
    order: bool = False,
    unsafe_hash: Literal[False] = False,
    frozen: Literal[False] = False,
    config: ConfigDict | type[object] | None = None,
    kw_only: bool = False,
    slots: Literal[False] = False,
) -> Callable[[type[T]], type[SyncDataclass]]
sync_dataclass(
    _cls: type[T],
    *,
    init: Literal[False] = False,
    repr: bool = True,
    eq: bool = True,
    order: bool = False,
    unsafe_hash: Literal[False] = False,
    frozen: Literal[False] = False,
    config: ConfigDict | type[object] | None = None,
    kw_only: bool = False,
    slots: Literal[False] = False,
) -> type[SyncDataclass]
sync_dataclass(
    _cls: type[T] | None = None,
    *,
    init: Literal[False] = False,
    repr: bool = True,
    eq: bool = True,
    order: bool = False,
    unsafe_hash: Literal[False] = False,
    frozen: Literal[False] = False,
    config: ConfigDict | type[object] | None = None,
    kw_only: bool = False,
    slots: Literal[False] = False,
) -> (
    Callable[[type[T]], type[SyncDataclass]]
    | type[SyncDataclass]
)

Make a reactive Pydantic dataclass.

Applies pydantic.dataclasses.dataclass and makes the class reactive, like a SyncModel. The class doesn't inherit from Reactive: check it with is_sync_dataclass. To make an existing dataclass reactive, subclass it.

Example:

from dataclasses import field

from syncwave import SyncList, Syncwave, sync_dataclass

syncwave = Syncwave()


@sync_dataclass
class Point:
    x: int = 0
    y: int = 0
    tags: SyncList[str] = field(default_factory=SyncList)


points = syncwave.create_store(SyncList[Point], name="points")
points.append(Point(x=1, y=2))
points[0].x = 3  # written to points.json


Usage Documentation

Reactive Dataclasses

Parameters:

Name Type Description Default
_cls type[T] | None

The class to decorate. Omit it to pass options: @sync_dataclass(...).

None
init Literal[False]

Must be False, as for pydantic.dataclasses.dataclass.

False
repr bool

Whether to generate __repr__.

True
eq bool

Whether to generate __eq__.

True
order bool

Whether to generate the comparison methods.

False
unsafe_hash Literal[False]

Must be False: reactive objects are mutable, so never hashable.

False
frozen Literal[False]

Must be False: a frozen dataclass cannot change, so it cannot sync.

False
config ConfigDict | type[object] | None

The Pydantic config.

None
kw_only bool

Whether the __init__ parameters are keyword-only.

False
slots Literal[False]

Must be False: reactive objects need a __dict__.

False

Returns:

Type Description
Callable[[type[T]], type[SyncDataclass]] | type[SyncDataclass]

The decorated class, or a decorator if _cls is omitted.

Source code in src/syncwave/sync_model.py
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
@dataclass_transform(field_specifiers=(field, Field, PrivateAttr))
def sync_dataclass(
    _cls: type[T] | None = None,
    *,
    init: Literal[False] = False,
    repr: bool = True,  # ruff: ignore[builtin-argument-shadowing]
    eq: bool = True,
    order: bool = False,
    unsafe_hash: Literal[False] = False,
    frozen: Literal[False] = False,
    config: ConfigDict | type[object] | None = None,
    kw_only: bool = False,
    slots: Literal[False] = False,
) -> F[[type[T]], type[SyncDataclass]] | type[SyncDataclass]:
    """Make a reactive Pydantic dataclass.

    Applies `pydantic.dataclasses.dataclass` and makes the class reactive, like a
    `SyncModel`. The class doesn't inherit from `Reactive`: check it with
    `is_sync_dataclass`. To make an existing dataclass reactive, subclass it.

    Example:
    ```python
    from dataclasses import field

    from syncwave import SyncList, Syncwave, sync_dataclass

    syncwave = Syncwave()


    @sync_dataclass
    class Point:
        x: int = 0
        y: int = 0
        tags: SyncList[str] = field(default_factory=SyncList)


    points = syncwave.create_store(SyncList[Point], name="points")
    points.append(Point(x=1, y=2))
    points[0].x = 3  # written to points.json
    ```

    ---

    Abstract: Usage Documentation
        [Reactive Dataclasses](https://syncwave.dev/usage/models/#reactive-dataclasses)

    Args:
        _cls: The class to decorate. Omit it to pass options: `@sync_dataclass(...)`.
        init: Must be `False`, as for `pydantic.dataclasses.dataclass`.
        repr: Whether to generate `__repr__`.
        eq: Whether to generate `__eq__`.
        order: Whether to generate the comparison methods.
        unsafe_hash: Must be `False`: reactive objects are mutable, so never hashable.
        frozen: Must be `False`: a frozen dataclass cannot change, so it cannot sync.
        config: The Pydantic config.
        kw_only: Whether the `__init__` parameters are keyword-only.
        slots: Must be `False`: reactive objects need a `__dict__`.

    Returns:
        The decorated class, or a decorator if `_cls` is omitted.

    """
    if unsafe_hash:
        raise TypeError("`unsafe_hash=True` is not supported.")
    if frozen:
        raise TypeError("`frozen=True` is not supported.")
    if slots:
        raise TypeError("`slots=True` is not supported.")

    def decorate(cls: type[T]) -> type[SyncDataclass]:
        if not isclass(cls):
            raise TypeError("`@sync_dataclass` must decorate a class.")
        if issubclass(cls, BaseModel):
            raise TypeError("`@sync_dataclass` cannot decorate a `BaseModel`.")
        if "__dataclass_fields__" in cls.__dict__:
            name = cls.__qualname__
            raise TypeError(
                f"`{name}` is already a dataclass. Remove the other decorator, or "
                f"subclass `{name}` to make a reactive version of it."
            )
        if "__slots__" in cls.__dict__:
            raise TypeError("`@sync_dataclass` cannot decorate a slotted class.")

        cs = SyncModel.__dict__["__get_pydantic_core_schema__"]
        cls.__get_pydantic_core_schema__ = cs  # ty: ignore[invalid-assignment]

        dc = py_dataclass(
            init=init,
            repr=repr,
            eq=eq,
            order=order,
            unsafe_hash=False,
            # frozen=False,  # may raise an unnecessary warning if set explicitly
            config=config,
            kw_only=kw_only,
            slots=False,
        )(cls)

        def get_dc_base_fn(fn_name: str) -> F:
            if fn_name in dc.__dict__:
                return dc.__dict__[fn_name]
            sw_attr = f"__syncwave_dc_base_{fn_name.strip('_')}__"
            if (fn := getattr(dc, sw_attr, None)) is not None:
                return fn
            return getattr(dc, fn_name)

        dc.__syncwave_dc_base_setattr__ = (b_setattr := get_dc_base_fn("__setattr__"))
        dc.__syncwave_dc_base_delattr__ = (b_delattr := get_dc_base_fn("__delattr__"))

        def update_(self: SyncDataclass, new: SyncDataclass) -> None:
            _update(self, new, b_setattr)

        @mut_reactive_op(inert_fn=b_setattr)
        def setattr_(self: SyncDataclass, name: str, value: Any) -> None:
            _setattr(self, name, value, b_setattr)

        @mut_reactive_op(inert_fn=b_delattr)
        def delattr_(self: SyncDataclass, name: str) -> None:
            _delattr(self, name, b_delattr)

        @reactive_op()
        def _copy(self: SyncDataclass) -> SyncDataclass:
            new = object.__new__(type(self))
            data = {k: v for k, v in self.__dict__.items() if k not in _LIVE_ATTRS}
            new.__dict__.update(data)
            _detach_fields(self, new)
            return new

        @reactive_op()
        def _deepcopy(self: SyncDataclass, memo: dict[int, Any]) -> SyncDataclass:
            new = object.__new__(type(self))
            data = {k: v for k, v in self.__dict__.items() if k not in _LIVE_ATTRS}
            new.__dict__.update(deepcopy(data, memo))
            return new

        base_repr = dc.__dict__["__repr__"]

        def _repr(self: SyncDataclass) -> str:
            return f"<{base_repr(self)} ({self.__syncwave_state__.value})>"

        dc.__syncwave_is_reactive__ = True
        dc.__syncwave_state__ = SyncState.INERT
        dc.sync_live = Reactive.__dict__["sync_live"]
        dc.sync_state = Reactive.__dict__["sync_state"]
        dc.__syncwave_init__ = _init
        dc.__syncwave_kill__ = _kill
        dc.__syncwave_update__ = update_
        dc.__getattribute__ = _getattribute
        dc.__setattr__ = setattr_
        dc.__delattr__ = delattr_
        dc.__copy__ = _copy
        dc.__deepcopy__ = _deepcopy
        if repr:
            dc.__repr__ = _repr
        dc.__hash__ = None

        if dc.__pydantic_complete__:
            _eager_validate(dc)
        return dc

    return decorate if _cls is None else decorate(_cls)

is_sync_dataclass

is_sync_dataclass(
    cls: type[Any],
) -> TypeIs[type[SyncDataclass]]

Check whether a class is a reactive dataclass, made with sync_dataclass.


Usage Documentation

Reactive Dataclasses

Parameters:

Name Type Description Default
cls type[Any]

The class to check.

required

Returns:

Type Description
TypeIs[type[SyncDataclass]]

True if cls is a reactive dataclass, False otherwise.

Source code in src/syncwave/sync_model.py
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
def is_sync_dataclass(cls: type[Any], /) -> TypeIs[type[SyncDataclass]]:
    """Check whether a class is a reactive dataclass, made with `sync_dataclass`.

    ---

    Abstract: Usage Documentation
        [Reactive Dataclasses](https://syncwave.dev/usage/models/#reactive-dataclasses)

    Args:
        cls: The class to check.

    Returns:
        `True` if `cls` is a reactive dataclass, `False` otherwise.

    """
    return isclass(cls) and is_pydantic_dataclass(cls) and is_reactive_cls(cls)