User-friendly tool for combining pycrdt for efficient concurrent content editing and pydantic for type safety and a pleasant developer experience.

Overview

• pymutantic.MutantModel - A type safe pycrdt.Doc ⟷ pydantic pydantic.BaseModel mapping with granular editing.
• pymutantic.JsonPathMutator - Make edits using json path.
• pymutantic.ModelVersionRegistry - Store a chain of versions for making granular schema migration edits.

Why do I want this?

The idea behind pymutantic is to provide views over the CRDT in the form of a pydantic model that you specify. There are two types of views:

• Read only snapshot: Inspect the state of the underlying CRDT with a frozen version of the pydantic model you specify. This model is read only, any changes are not reflected back to the CRDT (TODO: find a way to make this actually mutable)
• Mutable proxy: Make granular mutations to the data using a typed and mutable view over the underlying CRDT. Operations on this view are automatically synced with the underlying CRDT.

Installation

pip install pymutantic

Usage

MutantModel

Given a pydantic model...

from pydantic import BaseModel, Field
from typing import List

class Author(BaseModel):
    id: str
    name: str

class Comment(BaseModel):
    id: str
    author: Author
    content: str

class Post(BaseModel):
    id: str
    title: str
    content: str
    author: Author
    comments: List[Comment] = Field(default_factory=list)

class BlogPageConfig(BaseModel):
    collection: str
    posts: List[Post] = Field(default_factory=list)

Create pycrdt documents from instances of that model using the state constructor parameter:

from pymutantic import MutantModel

# Define the initial state
initial_state = BlogPageConfig(
    collection="tech",
    posts=[
        Post(
            id="post1",
            title="First Post",
            content="This is the first post.",
            author=Author(id="author1", name="Author One"),
            comments=[],
        )
    ]
)

# Create a CRDT document with the initial state
doc = MutantModel[BlogPageConfig](state=initial_state)

Get a snapshot (in the form of an instance of the pydantic model you specified) using the state property:

print(doc.snapshot)

BlogPageConfig(
    collection='tech',
    posts=[
        Post(
            id='post1',
            title='First Post',
            content='This is the first post.',
            author=Author(id='author1', name='Author One'),
            comments=[]
        )
    ]
)

NOTE: This is simply a snaphot any edits which are made to this copy are not reflected to the underlying CRDT.

Get a mutable view over the CRDT (in the form of an instance of the pydantic model you specified) and make granular edits using the mutate function

# Mutate the document
with doc.mutate() as state:
    state.posts[0].comments.append(Comment(
        id="comment1",
        author=Author(id="author2", name="Author Two"),
        content="Nice post!",
    ))
    state.posts[0].title = "First Post (Edited)"

print(doc.snapshot)

BlogPageConfig(
    collection='tech',
    posts=[
        Post(
            id='post1',
            title='First Post',
            content='This is the first post.',
            author=Author(id='author1', name='Author One'),
            comments=[
                Comment(
                    id="comment1",
                    author=Author(id="author2", name="Author Two"),
                    content="Nice post!",
                )
            ]
        )
    ]
)

NOTE: These edits are applied in bulk using a Doc.transaction

Type check your code to prevent errors:

empty_state = BlogPageConfig.model_validate({"collection": "empty", "posts": []})
doc = MutantModel[BlogPageConfig](state=empty_state)
doc.snapshot.psts

$ mypy . --check-untyped-defs --allow-redefinition

Use your IDE for a comfortable developer experience:

Get a binary update blob from the CRDT, for example for sending over the wire to other peers:

binary_update_blob: bytes = doc.update

Instantiate documents from a binary update blob (or multiple using the updates parameter which accepts a list of update blobs):

doc = MutantModel[BlogPageConfig](update=received_binary_update_blob)    

Apply more binary updates, by setting the update property:

doc.apply_updates(another_received_binary_update_blob)

JsonPathMutator

There is also a JsonPathMutator class which can be used to make edits to the document using json path:

# Mutate the document
from pymutantic import JsonPathMutator

with doc.mutate() as state:
    mutator = JsonPathMutator(state=state)
    mutator.set("$.posts[0].title", "Updated First Post")

print(doc.snapshot)

ModelVersionRegistry (experimental)

It is also possible to apply granular schema migration edits using the ModelVersionRegistry class. By storing multiple versions of a Model and implementing up and down functions (which in fact are making granular migrations) schema migrations can also be synchronized with other concurrent edits:

class ModelV1(BaseModel):
    schema_version: int = 1
    field: str
    some_field: str

    @classmethod
    def up(cls, state: typing.Any, new_state: typing.Any):
        raise NotImplementedError("can't migrate from null version")

    @classmethod
    def down(cls, state: typing.Any, new_state: typing.Any):
        raise NotImplementedError("can't migrate to null version")


class ModelV2(BaseModel):
    schema_version: int = 2
    some_field: str

    @classmethod
    def up(cls, state: ModelV1, new_state: "ModelV2"):
        del state.field

    @classmethod
    def down(cls, state: "ModelV2", new_state: ModelV1):
        new_state.field = "default"


from pymutantic import ModelVersionRegistry

migrate = ModelVersionRegistry([ModelV1, ModelV2])

doc = MutantModel[ModelV1](state=ModelV1(field="hello", some_field="world"))

# Make an independent edit
edit = MutantModel[ModelV1](update=doc.update)
with edit.mutate() as state:
    state.some_field = "earth"

# Migrate and apply the independent edit
doc = migrate(doc, to=ModelV2)
doc.update.apply_updates(edit.update)

ModelV2(schema_version=2, some_field='earth')