Add get_model_state to get validated doc.

pyproject.toml

@@ -44,6 +44,7 @@ test = [
     "mypy",
     "coverage[toml] >=7",
     "exceptiongroup; python_version<'3.11'",
+    "ruff>=0.13.3",
 ]
 docs = [
     "mkdocs",

python/pycrdt/_base.py

@@ -47,7 +47,6 @@ class BaseDoc:
     _txn_lock: threading.Lock
     _txn_async_lock: anyio.Lock
     _allow_multithreading: bool
-    _Model: Any
     _subscriptions: list[Subscription]
     _origins: dict[int, Any]
     _task_group: TaskGroup | None

python/pycrdt/_doc.py

@@ -2,7 +2,19 @@
 
 from functools import partial
 from inspect import iscoroutinefunction
-from typing import Any, Awaitable, Callable, Generic, Iterable, Literal, Type, TypeVar, Union, cast, overload
+from typing import (
+    Any,
+    Awaitable,
+    Callable,
+    Generic,
+    Iterable,
+    Literal,
+    Type,
+    TypeVar,
+    Union,
+    cast,
+    overload,
+)
 
 from anyio import BrokenResourceError, create_memory_object_stream
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
@@ -15,7 +27,9 @@
 from ._transaction import NewTransaction, ReadTransaction, Transaction
 
 T = TypeVar("T", bound=BaseType)
-TransactionOrSubdocsEvent = TypeVar("TransactionOrSubdocsEvent", bound=TransactionEvent | SubdocsEvent)
+TransactionOrSubdocsEvent = TypeVar(
+    "TransactionOrSubdocsEvent", bound=TransactionEvent | SubdocsEvent
+)
 
 
 class Doc(BaseDoc, Generic[T]):
@@ -35,7 +49,7 @@ def __init__(
         client_id: int | None = None,
         skip_gc: bool | None = None,
         doc: _Doc | None = None,
-        Model=None,
+        Model: Any | None = None,
         allow_multithreading: bool = False,
     ) -> None:
         """
@@ -47,8 +61,9 @@ def __init__(
             allow_multithreading: Whether to allow the document to be used in different threads.
         """
         super().__init__(
-            client_id=client_id, skip_gc=skip_gc, doc=doc, Model=Model, allow_multithreading=allow_multithreading
+            client_id=client_id, skip_gc=skip_gc, doc=doc, allow_multithreading=allow_multithreading
         )
+        self._Model = Model
         for k, v in init.items():
             self[k] = v
         if Model is not None:
@@ -150,6 +165,14 @@ def get_state(self) -> bytes:
             assert txn._txn is not None
             return self._doc.get_state(txn._txn)
 
+    def get_model_state(self) -> Any:
+        if self._Model is None:
+            raise RuntimeError(
+                "no Model defined for doc. Instantiate Doc with Doc(Model=PydanticModel)"
+            )
+        d = {k: self[k].to_py() for k in self._Model.model_fields}
+        return self._Model.model_validate(d)
+
     def get_update(self, state: bytes | None = None) -> bytes:
         """
         Args:
@@ -174,7 +197,7 @@ def apply_update(self, update: bytes) -> None:
             twin_doc.apply_update(update)
             d = {k: twin_doc[k].to_py() for k in self._Model.model_fields}
             try:
-                self._Model(**d)
+                self._Model.model_validate(d)
             except Exception as e:
                 self._twin_doc = Doc(dict(self))
                 raise e
@@ -292,7 +315,8 @@ def _roots(self) -> dict[str, T]:
 
     def observe(
         self,
-        callback: Callable[[TransactionEvent], None] | Callable[[TransactionEvent], Awaitable[None]],
+        callback: Callable[[TransactionEvent], None]
+        | Callable[[TransactionEvent], Awaitable[None]],
     ) -> Subscription:
         """
         Subscribes a callback to be called with the document change event.
@@ -405,7 +429,9 @@ async def main():
         observe = self.observe_subdocs if subdocs else self.observe
         if not self._send_streams[subdocs]:
             if async_transactions:
-                self._event_subscription[subdocs] = observe(partial(self._async_send_event, subdocs))
+                self._event_subscription[subdocs] = observe(
+                    partial(self._async_send_event, subdocs)
+                )
             else:
                 self._event_subscription[subdocs] = observe(partial(self._send_event, subdocs))
         send_stream, receive_stream = create_memory_object_stream[

python/pycrdt/_xml.py

@@ -269,9 +269,7 @@ def insert_embed(self, index: int, value: Any, attrs: dict[str, Any] | None = No
                 self._do_and_integrate("insert", value, txn._txn, index, _attrs)
             else:
                 # primitive type
-                self.integrated.insert_embed(
-                    txn._txn, index, value, _attrs
-                )
+                self.integrated.insert_embed(txn._txn, index, value, _attrs)
 
     def format(self, start: int, stop: int, attrs: dict[str, Any]) -> None:
         """

tests/test_doc.py

@@ -234,7 +234,7 @@ def test_get_update_exception():
 def test_apply_update_exception():
     doc = Doc()
     with pytest.raises(ValueError) as excinfo:
-        doc.apply_update(b"\xFF\xFF\xFF\xFF")
+        doc.apply_update(b"\xff\xff\xff\xff")
     assert "Cannot decode update" in str(excinfo.value)
 
 

tests/test_model.py

@@ -1,4 +1,4 @@
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import Tuple
 
 import pytest
@@ -52,3 +52,10 @@ class Delivery(BaseModel):
 
     assert str(local_doc["timestamp"]) == "2020-02-02T03:04:05Z"
     assert list(local_doc["dimensions"]) == ["10", "30"]
+
+    decoded = local_doc.get_model_state()
+    assert decoded.timestamp == datetime(2020, 2, 2, 3, 4, 5, tzinfo=timezone.utc)
+    assert decoded.dimensions == (
+        10,
+        30,
+    )

tests/test_transaction.py

@@ -1,5 +1,4 @@
 import gc
-import platform
 import sys
 import time
 from functools import partial