or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

array-operations.mdawareness.mddocument-management.mdindex.mdmap-operations.mdposition-undo.mdsynchronization.mdtext-operations.mdxml-support.md
tile.json

tessl/pypi-pycrdt

Python bindings for Yrs CRDT library providing collaborative data structures for real-time synchronization.

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/pycrdt@0.12.x

To install, run

npx @tessl/cli install tessl/pypi-pycrdt@0.12.0

index.mddocs/

pycrdt

Overview

pycrdt is a Python library that provides bindings for Yrs, a high-performance Conflict-free Replicated Data Type (CRDT) library written in Rust. It enables developers to build collaborative applications with automatic conflict resolution, supporting various data structures like arrays, maps, text, and XML documents. The library is designed for real-time collaborative editing, distributed systems, and offline-first applications that require eventual consistency across multiple clients.

Key features:

  • Collaborative data structures: Text, Array, Map, and XML with automatic conflict resolution
  • Real-time synchronization: Built-in support for document synchronization across clients
  • Type safety: TypedDoc, TypedArray, and TypedMap for runtime type checking
  • Event system: Comprehensive observation and event handling for all data types
  • Position tracking: StickyIndex for maintaining positions during concurrent edits
  • Undo/redo: Full undo manager with origin filtering and scope control
  • Async support: Full async/await support through anyio integration

Package Information

  • Package Name: pycrdt
  • Language: Python
  • Package Manager: pip
  • Installation: pip install pycrdt
  • Minimum Python Version: 3.9+
  • Build System: maturin (Python/Rust hybrid)

Core Imports

import pycrdt

# Import main document and transaction types
from pycrdt import Doc, TypedDoc
from pycrdt import Transaction, NewTransaction, ReadTransaction

# Import collaborative data types
from pycrdt import Text, Array, Map
from pycrdt import TypedArray, TypedMap
from pycrdt import XmlFragment, XmlElement, XmlText

# Import event types
from pycrdt import TextEvent, ArrayEvent, MapEvent, XmlEvent
from pycrdt import TransactionEvent, SubdocsEvent

# Import synchronization components
from pycrdt import Provider, Channel, Awareness
from pycrdt import Decoder, Encoder
from pycrdt import YMessageType, YSyncMessageType
from pycrdt import create_sync_message, handle_sync_message
from pycrdt import create_awareness_message, create_update_message
from pycrdt import read_message, write_message, write_var_uint

# Import utility types
from pycrdt import StickyIndex, Assoc, UndoManager
from pycrdt import Subscription, StackItem

# Import update functions
from pycrdt import get_state, get_update, merge_updates

# Import awareness utility
from pycrdt import is_awareness_disconnect_message

# Get package version
import pycrdt
print(pycrdt.__version__)  # Package version string

Basic Usage

import pycrdt
from pycrdt import Doc, Text, Array, Map

# Create a new collaborative document
doc = Doc()

# Create shared data structures
text = doc.get("shared_text", type=Text)
array = doc.get("shared_array", type=Array)
map_data = doc.get("shared_map", type=Map)

# Work with text
text.insert(0, "Hello, ")
text.insert(6, "world!")
print(str(text))  # "Hello, world!"

# Work with arrays
array.append("item1")
array.extend(["item2", "item3"])
print(len(array))  # 3

# Work with maps
map_data["key1"] = "value1"
map_data["key2"] = 42
print(map_data["key1"])  # "value1"

# Observe changes
def on_text_change(event):
    print(f"Text changed: {event.delta}")

subscription = text.observe(on_text_change)

# Make changes in a transaction
with doc.transaction() as txn:
    text += " How are you?"
    array.append("item4")
    map_data["timestamp"] = "2024-01-01"

Architecture

pycrdt follows a layered architecture:

  1. Document Layer: Doc serves as the container for all shared data types
  2. Transaction Layer: All mutations happen within Transaction contexts for atomicity
  3. Data Type Layer: Text, Array, Map, and XML types provide collaborative editing
  4. Event Layer: Comprehensive event system for observing changes
  5. Synchronization Layer: Provider and Channel handle network communication
  6. Awareness Layer: Client presence and metadata management

Capabilities

Document Management

Comprehensive document lifecycle management with transactions and type safety.

class Doc:
    def __init__(self, init: dict = {}, *, client_id: int | None = None) -> None
    def transaction(self, origin: Any = None) -> Transaction
    def get(self, key: str, *, type: type[T]) -> T
    def observe(self, callback: Callable[[TransactionEvent], None]) -> Subscription

See Document Management for complete API details.

Text Operations

Collaborative text editing with rich formatting and position tracking.

class Text:
    def insert(self, index: int, value: str, attrs: dict[str, Any] | None = None) -> None
    def format(self, start: int, stop: int, attrs: dict[str, Any]) -> None
    def diff(self) -> list[tuple[Any, dict[str, Any] | None]]
    def observe(self, callback: Callable[[TextEvent], None]) -> Subscription

See Text Operations for complete API details.

Array Operations

Collaborative arrays with list-like interface and change tracking.

class Array[T]:
    def append(self, value: T) -> None
    def extend(self, value: list[T]) -> None
    def insert(self, index: int, object: T) -> None
    def move(self, source_index: int, destination_index: int) -> None
    def observe(self, callback: Callable[[ArrayEvent], None]) -> Subscription

See Array Operations for complete API details.

Map Operations

Collaborative dictionaries with key-value storage and event tracking.

class Map[T]:
    def get(self, key: str, default_value: T_DefaultValue = None) -> T | T_DefaultValue | None
    def pop(self, key: str, default_value: T_DefaultValue = None) -> T | T_DefaultValue
    def update(self, value: dict[str, T]) -> None
    def observe(self, callback: Callable[[MapEvent], None]) -> Subscription

See Map Operations for complete API details.

XML Support

Structured XML document editing with elements, attributes, and text nodes.

class XmlElement:
    def __init__(self, tag: str | None = None, attributes: dict[str, str] | None = None) -> None
    @property
    def tag(self) -> str | None
    @property
    def attributes(self) -> XmlAttributesView
    @property
    def children(self) -> XmlChildrenView

See XML Support for complete API details.

Synchronization

Network synchronization with providers, channels, and message handling.

class Provider:
    def __init__(self, doc: Doc, channel: Channel, log: Logger | None = None) -> None
    async def start(self) -> None
    async def stop(self) -> None

def create_sync_message(ydoc: Doc) -> bytes
def handle_sync_message(message: bytes, ydoc: Doc) -> bytes | None

See Synchronization for complete API details.

Awareness Protocol

Client presence management and metadata sharing in collaborative sessions.

class Awareness:
    def __init__(self, ydoc: Doc, *, outdated_timeout: int = 30000) -> None
    def get_local_state(self) -> dict[str, Any] | None
    def set_local_state(self, state: dict[str, Any] | None) -> None
    def encode_awareness_update(self, client_ids: list[int]) -> bytes

See Awareness for complete API details.

Position Management & Undo

Persistent position tracking and comprehensive undo/redo operations.

class StickyIndex:
    def get_index(self, transaction: Transaction | None = None) -> int
    @classmethod
    def new(cls, sequence: Sequence, index: int, assoc: Assoc) -> Self

class UndoManager:
    def can_undo(self) -> bool
    def undo(self) -> bool
    def can_redo(self) -> bool
    def redo(self) -> bool

See Position Management & Undo for complete API details.

Common Patterns

Context Manager Pattern

All transactions use context managers for automatic cleanup:

with doc.transaction() as txn:
    # All mutations here are batched
    text.insert(0, "New content")
    array.append("new item")

Observer Pattern

All shared types support event observation:

def on_change(event):
    print(f"Change detected: {event}")

subscription = shared_type.observe(on_change)
# Remember to clean up
shared_type.unobserve(subscription)

Type Safety Pattern

Use typed variants for runtime type checking:

class MyDoc(TypedDoc):
    text_field: Text
    array_field: Array[str]
    map_field: Map[int]

doc = MyDoc()
doc.text_field.insert(0, "Hello")  # Type-safe access

Error Handling

pycrdt operations can raise the following exceptions:

  • ValueError: Invalid parameters or operations
  • TypeError: Type mismatches in typed variants
  • RuntimeError: Document or transaction state errors
try:
    with doc.transaction() as txn:
        text.insert(-1, "Invalid index")  # May raise ValueError
except ValueError as e:
    print(f"Operation failed: {e}")