PyCRDT

https://y-crdt.github.io/pycrdt/
Admin
PyCRDT is a Python CRDT library providing bindings for Yrs, enabling shared documents that...
Tokens:29,112
Snippets:348
Trust Score:9.7
Update:3 months ago
Show doc for...
Tokens:
Raw
### Set up Development Environment

Source: https://y-crdt.github.io/pycrdt/install

Create and activate a conda environment for development, installing pip and the Rust compiler.

```bash
micromamba create -n pycrdt-dev
micromamba activate pycrdt-dev
micromamba install pip rust
```

--------------------------------

### Install Pycrdt with pip

Source: https://y-crdt.github.io/pycrdt/install

Use this command to install the latest stable version of Pycrdt from PyPI.

```bash
pip install pycrdt
```

--------------------------------

### Install and Use Maturin for Rust Extensions

Source: https://y-crdt.github.io/pycrdt/install

Install the maturin tool once. Then, use 'maturin develop' to rebuild the Rust extension whenever Rust code changes, without needing to reinstall Pycrdt.

```bash
pip install maturin
maturin develop
```

--------------------------------

### Install Pycrdt with micromamba

Source: https://y-crdt.github.io/pycrdt/install

Install Pycrdt within an activated conda environment using micromamba.

```bash
micromamba install pycrdt
```

--------------------------------

### Async event iteration example for Text

Source: https://y-crdt.github.io/pycrdt/api_reference

Demonstrates how to use the `events` method with an async context manager and for-loop to process updates asynchronously. This pattern is essential for real-time collaboration.

```python
async def main():
    async with doc.events() as events:
        async for event in events:
            update: bytes = event.update
            ...
```

--------------------------------

### Using Provider with Lower-Level API

Source: https://y-crdt.github.io/pycrdt/api_reference

Demonstrates the lower-level API for managing a provider, including starting the provider task, waiting for it to be ready, and explicitly stopping it. This provides more control over the provider's lifecycle.

```python
task = asyncio.create_task(provider.start())
await provider.started.wait()
...
await provider.stop()
```

--------------------------------

### Create and Activate Conda Environment

Source: https://y-crdt.github.io/pycrdt/install

Before installing Pycrdt with micromamba, create a new environment named 'my_env' and activate it.

```bash
micromamba create -n my_env
micromamba activate my_env
```

--------------------------------

### `get(key, *, type)`

Source: https://y-crdt.github.io/pycrdt/api_reference

Retrieves a root type by key, casting it to a specified type if necessary, or creating a new one if it doesn't exist.

```APIDOC
## `get(key, *, type)`

Gets the document root type corresponding to the given key. If it already exists, it will be cast to the given type (if different), otherwise a new root type is created. 
```
doc.get("text", type=Text)
```

### Returns

| Type  | Description  |
|---|---|
| `T` | The root type corresponding to the given key, cast to the given type.  |

```

--------------------------------

### XmlText.format(start, stop, attrs)

Source: https://y-crdt.github.io/pycrdt/api_reference

Formats existing text within a specified range with given attributes.

```APIDOC
## XmlText.format(start, stop, attrs)

### Description
Formats existing text with attributes.

### Parameters
- `start` (int): The index at which to start applying the attributes (included).
- `stop` (int): The index at which to stop applying the attributes (excluded).
- `attrs` (dict[str, Any]): The attributes to apply.

### Method
`format(start, stop, attrs)`
```

--------------------------------

### Define and Use a Typed Map

Source: https://y-crdt.github.io/pycrdt/api_reference

Shows how to define a TypedMap with typed values and integrate it into a TypedDoc. Includes an example of accessing nested typed data.

```python
from pycrdt import Array, Map, TypedDoc, TypedMap

class MyMap(TypedMap):
    name: str
    toggle: bool
    nested: Array[bool]

class MyDoc(TypedDoc):
    map0: MyMap

doc = MyDoc()

doc.map0.name = "John"
doc.map0.toggle = True
doc.map0.nested = Array([True, False])

print(doc.map0._.to_py())
# {'nested': [True, False], 'toggle': True, 'name': 'John'}

```

--------------------------------

### format

Source: https://y-crdt.github.io/pycrdt/api_reference

Applies formatting attributes to a section of text between given start and stop indices.

```APIDOC
## format(start, stop, attrs)

### Description
Applies formatting attributes to a section of text between given start and stop indices.

### Parameters
#### Path Parameters
- **start** (int) - Required - The index at which to start applying attributes (included).
- **stop** (int) - Required - The index at which to stop applying attributes (excluded).
- **attrs** (dict[str, Any]) - Required - The formatting attributes to apply.
```

--------------------------------

### Install Pycrdt in Editable Mode

Source: https://y-crdt.github.io/pycrdt/install

Install Pycrdt in editable mode for development, including test and type groups. This command builds the Rust extension using maturin.

```bash
pip install -e . --group test --group types
```

--------------------------------

### TypedArray Usage Example

Source: https://y-crdt.github.io/pycrdt/api_reference

Demonstrates the usage of `TypedArray` for creating typed containers within a document, including nested structures and accessing elements.

```APIDOC
## TypedArray

Bases: `Typed`, `Generic[T]`

A container for an Array where values have types that can be other typed containers, e.g. a TypedMap. The subclass of `TypedArray[T]` must have a special `type: T` annotation where `T` is the same type. The underlying `Array` can be accessed with the special `_` attribute.

```python
from pycrdt import Array, TypedArray, TypedDoc, TypedMap

class MyMap(TypedMap):
    name: str
    toggle: bool
    nested: Array[bool]

class MyArray(TypedArray[MyMap]):
    type: MyMap

class MyDoc(TypedDoc):
    array0: MyArray

doc = MyDoc()

map0 = MyMap()
doc.array0.append(map0)
map0.name = "foo"
map0.toggle = True
map0.nested = Array([True, False])

print(doc.array0._.to_py())
# [{'name': 'foo', 'toggle': True, 'nested': [True, False]}]
print(doc.array0[0].name)
# foo
print(doc.array0[0].toggle)
# True
print(doc.array0[0].nested.to_py())
# [True, False]
```
```

--------------------------------

### Transaction with Context Manager

Source: https://y-crdt.github.io/pycrdt/api_reference

Use the `transaction` method with a context manager to create a new transaction or get the current one. If an origin is provided, it must match the origin of the current transaction.

```python
with doc.transaction():
    ...
```

--------------------------------

### Get value from Map with default

Source: https://y-crdt.github.io/pycrdt/api_reference

Retrieve a value by its key. If the key does not exist, return a specified default value or None. This prevents errors when a key might be absent.

```python
get(key: str) -> T | None
```

```python
get(key: str, default_value: T_DefaultValue) -> T | T_DefaultValue
```

--------------------------------

### Get item from Map

Source: https://y-crdt.github.io/pycrdt/api_reference

Access the value associated with a key using square bracket notation. This is the primary way to retrieve data from the map.

```python
Doc()["map0"] = map0 = Map({"foo": 0, "bar": 1})
assert map0["foo"] == 0
```

--------------------------------

### Accessing a Document Root Type

Source: https://y-crdt.github.io/pycrdt/api_reference

Use dictionary-like access to retrieve a root type from the document by its key. This is useful for getting existing shared types within the document.

```python
text = doc["text"]

```

--------------------------------

### Getting a Document Root Type with Type Casting

Source: https://y-crdt.github.io/pycrdt/api_reference

Retrieve a root type by key and optionally cast it to a specific type. If the root type does not exist, it will be created. This is useful for ensuring type safety when accessing document elements.

```python
doc.get("text", type=Text)

```

--------------------------------

### `__init__(init={}, *, client_id=None, skip_gc=None, doc=None, Model=None, allow_multithreading=False)`

Source: https://y-crdt.github.io/pycrdt/api_reference

Initializes a new `Doc` instance. Allows setting initial root types, client ID, and garbage collection behavior.

```APIDOC
## `__init__(init={}, *, client_id=None, skip_gc=None, doc=None, Model=None, allow_multithreading=False)`

### Parameters

| Name  | Type  | Description  | Default  |
|---|---|---|---|
| `init` | `dict[str, T]` | The initial root types of the document. | `{}` |
| `client_id` | `int | None` | An optional client ID for the document. | `None` |
| `skip_gc` | `bool | None` | Whether to skip garbage collection on deleted collections on transaction commit. | `None` |
| `allow_multithreading` | `bool` | Whether to allow the document to be used in different threads. | `False` |

```

--------------------------------

### Using Provider with Async Context Manager

Source: https://y-crdt.github.io/pycrdt/api_reference

Illustrates the recommended way to use a provider for synchronizing a document through a channel, utilizing an async context manager. This ensures proper startup and shutdown.

```python
async with provider:
    ...
```

--------------------------------

### Map.__init__(init=None, *, _doc=None, _integrated=None)

Source: https://y-crdt.github.io/pycrdt/api_reference

Initializes a new Map instance, optionally with an initial dictionary of key-value pairs.

```APIDOC
## Map.__init__(init=None, *, _doc=None, _integrated=None)

### Description
Creates a map with an optional initial value.

### Parameters
#### Path Parameters
- **init** (dict[str, T] | None) - Optional - The dictionary from which to initialize the map. Defaults to None.
- **_doc** (any) - Optional - Internal parameter.
- **_integrated** (any) - Optional - Internal parameter.
```

--------------------------------

### Getting the Length of an Array

Source: https://y-crdt.github.io/pycrdt/api_reference

The `__len__` method returns the number of items currently stored in the Array.

```python
Doc()["array"] = array = Array([2, 3, 0])
assert len(array) == 3
```

--------------------------------

### Array.__init__

Source: https://y-crdt.github.io/pycrdt/api_reference

Initializes a new Array instance, optionally with an initial list of elements.

```APIDOC
## Array.__init__

### Description
Creates an array with an optional initial value.

### Method
__init__

### Parameters
#### Parameters
- **init** (list[T] | None) - Optional - The list from which to initialize the array. Defaults to None.

### Example
```python
array0 = Array()
array1 = Array(["foo", 3, array0])
```
```

--------------------------------

### Provider Initialization and Usage

Source: https://y-crdt.github.io/pycrdt/api_reference

Details the initialization and usage patterns for the Provider class, which synchronizes a document through a channel.

```APIDOC
## Provider

### `__init__(doc, channel, log=None)`

#### Description
A provider synchronizes a document through a channel.

#### Usage
The provider should preferably be used with an async context manager:

```python
async with provider:
    ...
```

However, a lower-level API can also be used:

```python
task = asyncio.create_task(provider.start())
await provider.started.wait()
...
await provider.stop()
```

#### Parameters

- **doc** (Doc): The `Doc` to connect through the `Channel`. (required)
- **channel** (Channel): The `Channel` through which to connect the `Doc`. (required)
- **log** (Logger | None): An optional logger. (default: None)
```

```APIDOC
### `start(*, task_status=TASK_STATUS_IGNORED)` `async`

#### Description
Start the provider.

#### Parameters

- **task_status** (TaskStatus[None]): The status to set when the task has started. (default: TASK_STATUS_IGNORED)
```

```APIDOC
### `stop()` `async`

#### Description
Stop the provider.
```

--------------------------------

### Get Map keys

Source: https://y-crdt.github.io/pycrdt/api_reference

Obtain an iterable of all keys present in the map. This is useful for operations that only require the keys.

```python
keys() -> Iterable[str]
```

--------------------------------

### Initialize Decoder with a byte stream

Source: https://y-crdt.github.io/pycrdt/api_reference

Create a Decoder instance by providing the byte stream to read from.

```python
decoder = Decoder(stream)
```

--------------------------------

### Get Map items

Source: https://y-crdt.github.io/pycrdt/api_reference

Obtain an iterable of key-value pairs from the map. This is useful for iterating through all entries simultaneously.

```python
items() -> Iterable[tuple[str, T]]
```

--------------------------------

### StackItem.__init__

Source: https://y-crdt.github.io/pycrdt/api_reference

Creates a new StackItem. This constructor initializes a unit of work for the UndoManager with document, deletions, insertions, and optional metadata.

```APIDOC
## StackItem.__init__(doc, deletions, insertions, meta=None)

### Description
Create a new StackItem.

### Method
`constructor`

### Parameters
#### Path Parameters
- **doc** (Doc) - Required - The document this stack item belongs to.
- **deletions** (IdSet) - Required - The IdSet of deletions.
- **insertions** (IdSet) - Required - The IdSet of insertions.
- **meta** (MetaT | None) - Optional - Optional metadata (can be any Python object). Defaults to None.
```

--------------------------------

### Generate messages from the decoder stream

Source: https://y-crdt.github.io/pycrdt/api_reference

Use read_messages() to get a generator that yields all messages from the byte stream.

```python
for message in decoder.read_messages():
    ...
```

--------------------------------

### Text.__init__

Source: https://y-crdt.github.io/pycrdt/api_reference

Initializes a new Text object with an optional initial string value.

```APIDOC
## Text.__init__(init=None, *, _doc=None, _integrated=None)

### Description
Creates a text with an optional initial value.

### Method
`__init__`

### Parameters
#### Path Parameters
- **init** (str | None) - Optional - The string from which to initialize the text. Defaults to None.

### Request Example
```python
text = Text("Hello, World!")
```

```

--------------------------------

### Initialize Map

Source: https://y-crdt.github.io/pycrdt/api_reference

Create a new Map instance. It can be initialized empty or with a dictionary of key-value pairs. Nested maps are also supported.

```python
map0 = Map()
map1 = Map({"foo": 0, "bar": 3, "baz": map0})
```

--------------------------------

### Read a single message from the decoder stream

Source: https://y-crdt.github.io/pycrdt/api_reference

Use read_message() to get the current message from the byte stream.

```python
message = decoder.read_message()
```

--------------------------------

### Initialize Text object

Source: https://y-crdt.github.io/pycrdt/api_reference

Create a new Text object, optionally initializing it with a given string. If no initial value is provided, the Text object will be empty.

```python
text = Text("Hello, World!")
```

--------------------------------

### String representation of Map

Source: https://y-crdt.github.io/pycrdt/api_reference

Get a string representation of the map, which is formatted as a JSON object. This is useful for debugging or logging.

```python
Doc()["map0"] = map0 = Map({"foo": 0, "bar": 1})
assert str(map0) == '{"foo":0,"bar":1}'
```

--------------------------------

### Initializing an Array

Source: https://y-crdt.github.io/pycrdt/api_reference

An Array can be initialized empty or with an initial list of values. The `__init__` method handles this.

```python
array0 = Array()
array1 = Array(["foo", 3, array0])
```

--------------------------------

### Iterate over Map keys

Source: https://y-crdt.github.io/pycrdt/api_reference

Iterate directly over a Map object to get its keys. This allows for processing each key-value pair within the map.

```python
Doc()["map0"] = map0 = Map({"foo": 0, "bar": 0})
for key in m0:
    assert map[key] == 0
```

--------------------------------

### XmlElement Initialization

Source: https://y-crdt.github.io/pycrdt/api_reference

Creates a new XmlElement with optional tag, attributes, and contents.

```APIDOC
## XmlElement Initialization

### `__init__(tag=None, attributes=None, contents=None, *, _doc=None, _integrated=None)`
Creates an XML element.

Parameters:
Name | Type | Description | Default  
---|---|---|---  
`tag` |  `str | None` |  The tag of the element (required). |  `None`  
`attributes` |  `dict[str, str] | Iterable[tuple[str, str]] | None` |  The optional attributes of the element. |  `None`  
`contents` |  `Iterable[XmlFragment | XmlElement | XmlText] | None` |  The optional contents of the element. |  `None`
```

--------------------------------

### new_transaction

Source: https://y-crdt.github.io/pycrdt/api_reference

Creates a new transaction. This method ensures that no transaction is ongoing before starting a new one, with options for asynchronous context management and timeouts.

```APIDOC
## `new_transaction(origin=None, timeout=None)`

### Description
Creates a new transaction. Unlike `transaction()`, this method will not reuse an ongoing transaction. If there is already an ongoing transaction, this method will wait (with an optional timeout) until the current transaction has finished.

### Parameters
#### Path Parameters
- **origin** (Any) - Optional - An optional origin to set on this transaction.
- **timeout** (float | None) - Optional - An optional timeout (in seconds) to acquire a new transaction.

### Raises
- **RuntimeError**: Already in a transaction.
- **TimeoutError**: Could not acquire transaction.

### Returns
- **NewTransaction**: A new transaction.

### Usage
Use an async context manager:
```python
async with doc.new_transaction():
    ...
```
Or a sync context manager:
```python
with doc.new_transaction():
    ...
```
```

--------------------------------

### NewTransaction Usage

Source: https://y-crdt.github.io/pycrdt/api_reference

Demonstrates how to use the NewTransaction class for mutating a document using context managers.

```APIDOC
## NewTransaction

### Description
A read-write transaction that can be used to mutate a document.

### Usage
It can be used with a context manager or an async context manager (see Doc.new_transaction()):

```python
with doc.new_transaction():
    ...

async with doc.new_transaction():
    ...
```

### Origin Property

- **origin** (property): The origin of the transaction.

  - **Raises**:
    - RuntimeError: No current transaction.
```

--------------------------------

### `from_snapshot(snapshot, doc)`

Source: https://y-crdt.github.io/pycrdt/api_reference

A static method to create a new `Doc` instance restored from a given snapshot, using an existing `Doc` for configuration.

```APIDOC
## `from_snapshot(snapshot, doc)` `staticmethod`

Create a new Doc from a Snapshot and an original Doc. Args: snapshot: The Snapshot to restore to. doc: The original Doc to use for options/state. Returns: A new Doc instance restored to the snapshot state.

```

--------------------------------

### Awareness Class Methods

Source: https://y-crdt.github.io/pycrdt/api_reference

Provides methods to manage client awareness states, including initialization, updating, encoding, and observing changes.

```APIDOC
## Awareness

### Description
Manages client awareness states and metadata for collaborative editing.

### Methods

#### `__init__(ydoc, *, outdated_timeout=30000)`
Initializes the Awareness object.

- **Parameters**:
  - `ydoc` (Doc): The Doc to associate the awareness with.
  - `outdated_timeout` (int): The timeout in milliseconds to consider a client gone. Defaults to 30000.

#### `apply_awareness_update(update, origin)`
Applies a binary awareness update and notifies subscribers.

- **Parameters**:
  - `update` (bytes): The binary update to apply.
  - `origin` (Any): The origin of the update.

#### `encode_awareness_update(client_ids)`
Encodes an awareness update for the specified client IDs.

- **Parameters**:
  - `client_ids` (list[int]): A list of client IDs for which to create an update.
- **Returns**:
  - `bytes`: The encoded awareness update.

#### `get_local_state()`
Retrieves the local client state.

- **Returns**:
  - `dict[str, Any] | None`: The local state, if any.

#### `observe(callback)`
Registers a callback function to be called when awareness changes occur.

- **Parameters**:
  - `callback` (Callable[[str, tuple[dict[str, Any], Any]], None]): The callback function to execute.
- **Returns**:
  - `str`: The subscription ID that can be used to unobserve.

#### `remove_awareness_states(client_ids, origin)`
Removes awareness states for the specified client IDs.

- **Parameters**:
  - `client_ids` (list[int]): A list of client IDs for which to remove awareness states.
  - `origin` (Any): The origin of the update.

#### `set_local_state(state)`
Updates the local client state and meta, and notifies subscribers.

- **Parameters**:
  - `state` (dict[str, Any] | None): The new local state.

#### `set_local_state_field(field, value)`
Sets a specific field in the local client state.

- **Parameters**:
  - `field` (str): The field of the local state to set.
  - `value` (Any): The value for the specified field.

#### `start(*, task_status=TASK_STATUS_IGNORED)`
Starts the periodic awareness update process.

#### `stop()`
Stops the periodic awareness update process.

#### `unobserve(id)`
Unregisters a subscription ID from awareness change notifications.

- **Parameters**:
  - `id` (str): The subscription ID to unregister.
```

--------------------------------

### Formatting Text in XmlText

Source: https://y-crdt.github.io/pycrdt/api_reference

Formats a specified range of text within an XmlText object with given attributes. Ensure start and stop indices are valid.

```python
format(start, stop, attrs)
```

--------------------------------

### Get the length of Text

Source: https://y-crdt.github.io/pycrdt/api_reference

Determine the number of characters in the Text object using the `len()` function. This returns an integer representing the total character count.

```python
Doc()["text"] = text = Text("Hello")
assert len(text) == 5
```

--------------------------------

### Define and Use a Typed Document

Source: https://y-crdt.github.io/pycrdt/api_reference

Demonstrates how to define a TypedDoc with typed root shared values and interact with its underlying Doc.

```python
from pycrdt import Array, Doc, Map, Text, TypedDoc

class MyDoc(TypedDoc):
    map0: Map[int]
    array0: Array[bool]
    text0: Text

doc = MyDoc()

doc.map0["foo"] = 3
doc.array0.append(True)
doc.text0 += "Hello"
untyped_doc: Doc = doc._

```

--------------------------------

### Get Map length

Source: https://y-crdt.github.io/pycrdt/api_reference

Use the 'len()' function to determine the number of key-value pairs currently stored in the map. This is useful for size checks or loop bounds.

```python
Doc()["map0"] = map0 = Map({"foo": 0, "bar": 1})
assert len(map0) == 2
```

--------------------------------

### Nested Transactions

Source: https://y-crdt.github.io/pycrdt/usage

Transactions can be nested. Changes made in inner transactions are applied to the outer transaction. This example shows changes being made within nested contexts.

```python
with doc.transaction() as t0:
    text0 += ", World!"
    with doc.transaction() as t1:
        array0.append("bar")
        with doc.transaction() as t2:
            map0["key1"] = "value1"

```

--------------------------------

### observe

Source: https://y-crdt.github.io/pycrdt/api_reference

Subscribes a callback to be called with the text event.

```APIDOC
## observe(callback)

### Description
Subscribes a callback to be called with the text event.

### Parameters
#### Path Parameters
- **callback** (Callable[[TextEvent], None]) - Required - The callback to call with the TextEvent.
```

--------------------------------

### Run Tests with Pytest

Source: https://y-crdt.github.io/pycrdt/install

Execute the project's tests using the pytest framework.

```bash
pytest -v
```

--------------------------------

### Generate and Apply Document Updates

Source: https://y-crdt.github.io/pycrdt/usage

Generate binary updates from document changes and apply them to a remote document to synchronize states.

```python
update = doc.get_update()

# the (binary) update could travel on the wire to a remote machine:

remote_doc = Doc()
remote_doc.apply_update(update)

remote_doc["text0"] = text0 = Text()
remote_doc["array0"] = array0 = Array()
remote_doc["map0"] = map0 = Map()

```

--------------------------------

### Prepending Items to an Array

Source: https://y-crdt.github.io/pycrdt/api_reference

Use the `__radd__` method for prepending a list of items to the beginning of an Array. This operation returns a new Array.

```python
Doc()["array"] = array = Array(["bar", "baz"])
array = ["foo"] + array
assert array.to_py() == ["foo", "bar", "baz"]
```

--------------------------------

### Map.get(*args)

Source: https://y-crdt.github.io/pycrdt/api_reference

Retrieves a value by key, returning a default value or None if the key is not found.

```APIDOC
## Map.get(*args)

### Description
Returns the value corresponding to the given key if it exists, otherwise returns the default value if passed, or `None`.

### Parameters
#### Path Parameters
- **args** - The key of the value to get, and an optional default value.

### Returns
- The value at the given key, or the default value or `None`.
```

--------------------------------

### Get characters from Text by index or slice

Source: https://y-crdt.github.io/pycrdt/api_reference

Retrieve a portion of the Text object using an integer index or a slice. This operation returns the specified characters as a string without modifying the original Text object.

```python
Doc()["text"] = text = Text("Hello, World!")
assert text[:5] == "Hello"
```

--------------------------------

### Apply Binary Update to Document

Source: https://y-crdt.github.io/pycrdt/usage

Use this to apply a binary update received from a network source, such as a WebSocket, to a shared document.

```python
remote_doc.apply_update(update)
```

--------------------------------

### Asynchronous Context Manager for Transactions

Source: https://y-crdt.github.io/pycrdt/usage

Use `async with doc.transaction():` to manage transactions in an asynchronous environment. This allows for asynchronous callbacks, preventing the event loop from being blocked by document changes.

```python
async def async_callback(event):
    await send(event.update)

doc.observe(async_callback)

async with doc.transaction():
    ...

```

--------------------------------

### to_py

Source: https://y-crdt.github.io/pycrdt/api_reference

Returns the text as a Python string.

```APIDOC
## to_py()

### Description
Returns the text as a Python string.

### Returns
- **str | None** - The text as a Python `str`.
```

--------------------------------

### Async Transaction with new_transaction

Source: https://y-crdt.github.io/pycrdt/api_reference

Use `new_transaction` with an async context manager for accessing the document in the same thread. Ensure the Doc is created with `allow_multithreading=False`.

```python
async with doc.new_transaction():
    ...
```

--------------------------------

### Asynchronously Iterating Over Document Events

Source: https://y-crdt.github.io/pycrdt/api_reference

This snippet demonstrates how to asynchronously iterate over document events using an async context manager and an async for-loop. It's suitable for real-time event handling in asynchronous applications.

```python
async def main():
    async with doc.events() as events:
        async for event in events:
            update: bytes = event.update
            ...

```

--------------------------------

### observe

Source: https://y-crdt.github.io/pycrdt/api_reference

Subscribes a callback function to be invoked whenever the document changes.

```APIDOC
## `observe(callback)`

### Description
Subscribes a callback to be called with the document change event. If the callback is asynchronous, async transactions must be used.

### Parameters
#### Path Parameters
- **callback** (Callable[[TransactionEvent], None] | Callable[[TransactionEvent], Awaitable[None]]) - Required - The callback to call with the TransactionEvent.

### Returns
- **Subscription**: The subscription that can be used to unobserve().
```

--------------------------------

### StickyIndex.new

Source: https://y-crdt.github.io/pycrdt/api_reference

Creates a new sticky index positioned before or after a specified index within a sequence. This class method is used to establish a new sticky index with a defined association.

```APIDOC
## StickyIndex.new

### Description
Create a sticky index before or after the specified index.

### Method
`classmethod`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
Name | Type | Description | Default  
---|---|---|---
`sequence` | `Sequence` | The Array or Text to get the sticky index from. | _required_
`index` | `int` | The index at which the sticky index should remain. | _required_
`assoc` | `Assoc` | The Assoc the sticky index should be associated with (before or after). | _required_

### Request Example
None

### Response
#### Success Response (200)
Type | Description  
---|---
`Self` | The sticky index.

#### Response Example
None
```

--------------------------------

### Asynchronously Iterate Over Shared Data Events

Source: https://y-crdt.github.io/pycrdt/usage

In an asynchronous environment, you can iterate over events directly using `async with text0.events() as events:`. This avoids the need for explicit callback registration.

```python
async def main():
    async with text0.events() as events:
        async for event in events:
            # process the event


```

--------------------------------

### Clone Pycrdt Repository

Source: https://y-crdt.github.io/pycrdt/install

Clone the Pycrdt repository from GitHub to set up a development environment.

```bash
git clone https://github.com/y-crdt/pycrdt.git
cd pycrdt
```

--------------------------------

### Observe Document Changes for Synchronization

Source: https://y-crdt.github.io/pycrdt/usage

Observe document-level changes using `doc.observe`. The callback receives a `TransactionEvent` containing the binary `update` data, which can be sent over a network to synchronize other documents.

```python
from pycrdt import TransactionEvent

def handle_doc_changes(event: TransactionEvent):
    update: bytes = event.update
    # send binary update on the wire

doc.observe(handle_doc_changes)

```

--------------------------------

### Asynchronously iterate over Map events

Source: https://y-crdt.github.io/pycrdt/api_reference

Use an async context manager and for-loop to process events from the map. A buffer can be configured to handle events at a different rate than they are produced.

```python
async def main():
    async with doc.events() as events:
        async for event in events:
            update: bytes = event.update
            ...
```

```python
events(deep: Literal[False], max_buffer_size: float = float('inf')) -> MemoryObjectReceiveStream[BaseEvent]
```

```python
events(deep: Literal[True], max_buffer_size: float = float('inf')) -> MemoryObjectReceiveStream[list[BaseEvent]]
```

--------------------------------

### Array.__radd__

Source: https://y-crdt.github.io/pycrdt/api_reference

Prepends a list of items to the beginning of the array.

```APIDOC
## Array.__radd__

### Description
Prepends a list of items to the array.

### Method
__radd__

### Parameters
#### Parameters
- **value** (list[T]) - Required - The list of items to prepend.

### Returns
#### Returns
- **Array[T]** - The prepended array.

### Example
```python
Doc()["array"] = array = Array(["bar", "baz"])
array = ["foo"] + array
assert array.to_py() == ["foo", "bar", "baz"]
```
```

--------------------------------

### `observe(callback)`

Source: https://y-crdt.github.io/pycrdt/api_reference

Subscribes a callback function to be executed whenever an event occurs on the array. The callback receives an ArrayEvent object.

```APIDOC
## `observe(callback)`

### Description
Subscribes a callback to be called with the array event.

### Parameters
#### Path Parameters
- None

#### Query Parameters
- None

#### Request Body
- **callback** (Callable[[ArrayEvent], None]) - Required - The callback to call with the ArrayEvent.

### Request Example
```json
{
  "example": "observe(myCallbackFunction)"
}
```

### Response
#### Success Response (200)
- **subscription** (Subscription) - A subscription object that can be used to unobserve.

#### Response Example
```json
{
  "example": "<subscription_object>"
}
```
```

--------------------------------

### create_sync_message

Source: https://y-crdt.github.io/pycrdt/api_reference

Creates a SYNC_STEP1 message containing the state of a Ypy Doc.

```APIDOC
## create_sync_message(ydoc)

### Description
Creates a SYNC_STEP1 message that contains the state of a Doc.

### Parameters
#### Path Parameters
* None

#### Query Parameters
* None

#### Request Body
* **ydoc** (Doc) - Required - The Doc for which to create the message.

### Request Example
```json
{
  "example": "request body"
}
```

### Response
#### Success Response (200)
* **bytes** - A SYNC_STEP1 message.

#### Response Example
```json
{
  "example": "response body"
}
```
```

--------------------------------

### XmlFragment.events

Source: https://y-crdt.github.io/pycrdt/api_reference

Allows asynchronous iteration over the shared type events with optional deep event tracking and buffer size limits.

```APIDOC
## `events(deep=False, max_buffer_size=float('inf'))`

Allows to asynchronously iterate over the shared type events, without using a callback. A buffer is used to store the events, allowing to iterate at a (temporarily) slower rate than they are produced.
This method must be used with an async context manager and an async for-loop:
```python
async def main():
    async with doc.events() as events:
        async for event in events:
            update: bytes = event.update
            ...
```

Parameters:
Name | Type | Description | Default  
---|---|---|---
`deep` |  `bool` |  Whether to iterate over the nested events. |  `False`  
`max_buffer_size` |  `float` |  Maximum number of events that can be buffered. |  `float('inf')`  
Returns:
Type | Description  
---|---
|  An async iterator over the shared type events.
```

--------------------------------

### Array.__add__

Source: https://y-crdt.github.io/pycrdt/api_reference

Extends the array with a list of items. This method allows for appending multiple elements at once.

```APIDOC
## Array.__add__

### Description
Extends the array with a list of items.

### Method
__add__

### Parameters
#### Parameters
- **value** (list[T]) - Required - The items that will extend the array.

### Returns
#### Returns
- **Array[T]** - The extended array.

### Example
```python
Doc()["array"] = array = Array(["foo"])
array += ["bar", "baz"]
assert array.to_py() == ["foo", "bar", "baz"]
```
```

--------------------------------

### Encoder.write_var_string(text)

Source: https://y-crdt.github.io/pycrdt/api_reference

Encodes a given string into the binary stream.

```APIDOC
## Encoder.write_var_string(text)

### Description
Encodes a string into the binary stream.

### Parameters
#### Path Parameters
- **text** (str) - Required - The string to encode.
```

--------------------------------

### Asynchronous Document Event Handling

Source: https://y-crdt.github.io/pycrdt/usage

In an asynchronous environment, iterate over document events to process updates as they occur. This is an alternative to registering callbacks.

```python
async def main():
    async with doc.events() as events:
        async for event in events:
            update: bytes = event.update
            # send binary update on the wire
```

--------------------------------

### Map.__str__()

Source: https://y-crdt.github.io/pycrdt/api_reference

Returns a string representation of the map.

```APIDOC
## Map.__str__()

### Description
Returns the string representation of the map.

### Returns
- **str**: The string representation of the map.
```

--------------------------------

### Sync Transaction with new_transaction

Source: https://y-crdt.github.io/pycrdt/api_reference

Use `new_transaction` with a sync context manager when the ongoing transaction must run in another thread to avoid deadlocks. Ensure the Doc is created with `allow_multithreading=True`.

```python
with doc.new_transaction():
    ...
```

--------------------------------

### Text.__str__

Source: https://y-crdt.github.io/pycrdt/api_reference

Returns the Text object as a standard Python string.

```APIDOC
## Text.__str__()

### Description
Returns the text as a Python `str`.

### Method
`__str__`

### Response
#### Success Response
- **str** - The text as a Python `str`.

```

--------------------------------

### `__getitem__(key)`

Source: https://y-crdt.github.io/pycrdt/api_reference

Retrieves a root type from the document using its key. This allows direct access to shared types within the document.

```APIDOC
## `__getitem__(key)`

Gets the document root type corresponding to the given key:
```
text = doc["text"]
```

### Parameters

| Name  | Type  | Description  | Default  |
|---|---|---|---|
| `key` | `str` | The key of the root type to get. | _required_ |

### Returns

| Type  | Description  |
|---|---|
| `T` | The document root type.  |

```

--------------------------------

### Text.__getitem__

Source: https://y-crdt.github.io/pycrdt/api_reference

Retrieves characters from the Text object at a specified index or slice.

```APIDOC
## Text.__getitem__(key)

### Description
Gets the characters at the given index or slice.

### Method
`__getitem__`

### Parameters
#### Path Parameters
- **key** (int | slice) - Required - The index or slice of the characters to retrieve.

### Response
#### Success Response
- **str** - The characters at the given index or slice.

### Request Example
```python
Doc()["text"] = text = Text("Hello, World!")
assert text[:5] == "Hello"
```

```

--------------------------------

### Text.events

Source: https://y-crdt.github.io/pycrdt/api_reference

Allows asynchronous iteration over the shared type events with optional buffering.

```APIDOC
## Text.events(deep=False, max_buffer_size=float('inf'))

### Description
Allows to asynchronously iterate over the shared type events, without using a callback. A buffer is used to store the events, allowing to iterate at a (temporarily) slower rate than they are produced.
This method must be used with an async context manager and an async for-loop.

### Method
`events`

### Parameters
#### Path Parameters
- **deep** (bool) - Optional - Whether to iterate over the nested events. Defaults to False.
- **max_buffer_size** (float) - Optional - Maximum number of events that can be buffered. Defaults to float('inf').

### Response
#### Success Response
- **An async iterator over the shared type events.**

### Request Example
```python
async def main():
    async with doc.events() as events:
        async for event in events:
            update: bytes = event.update
            ...
```

```

--------------------------------

### Array.__str__

Source: https://y-crdt.github.io/pycrdt/api_reference

Returns a string representation of the array.

```APIDOC
## Array.__str__

### Description
Returns the string representation of the array.

### Method
__str__

### Returns
#### Returns
- **str** - The string representation of the array.

### Example
```python
Doc()["array"] = array = Array([2, 3, 0])
assert str(array) == "[2,3,0]"
```
```

--------------------------------

### Snapshot.from_doc

Source: https://y-crdt.github.io/pycrdt/api_reference

Creates a snapshot from a document. This class method captures the current state of a document as a snapshot.

```APIDOC
## Snapshot.from_doc(doc)

### Description
Create a snapshot from a document.

### Method
`classmethod`

### Parameters
#### Path Parameters
- **doc** ('Doc') - Required - The document to snapshot.

### Returns
#### Success Response
- **Snapshot** - The snapshot of the document's current state.
```

--------------------------------

### Asynchronous Transactions with `new_transaction()`

Source: https://y-crdt.github.io/pycrdt/usage

When used with an async context manager, `new_transaction()` yields to the event loop until a transaction is acquired. This is designed for asynchronous programming environments.

```python
from anyio import create_task_group, run
from pycrdt import Doc

doc = Doc()

async def create_new_transaction():
    async with doc.new_transaction(timeout=3):
        ...

async def main():
    async with create_task_group() as tg:
        tg.start_soon(create_new_transaction)
        tg.start_soon(create_new_transaction)

run(main)

```

--------------------------------

### `keys()`

Source: https://y-crdt.github.io/pycrdt/api_reference

Returns an iterable of the names (keys) of all root types currently defined in the document.

```APIDOC
## `keys()`

### Returns

| Type  | Description  |
|---|---|
| `Iterable[str]` | An iterable over the names of the document root types.  |

```

--------------------------------

### XmlText.diff()

Source: https://y-crdt.github.io/pycrdt/api_reference

Returns a list of formatted chunks representing the current text content.

```APIDOC
## XmlText.diff()

### Description
Returns a list of formatted chunks that the current text corresponds to. Each list item is a tuple containing the chunk's contents and formatting attributes. The contents is usually the text as a string, but may be other data for embedded objects.

### Returns
- `list[tuple[Any, dict[str, Any] | None]]`: A list of formatted chunks.
```

--------------------------------

### Text.diff

Source: https://y-crdt.github.io/pycrdt/api_reference

Returns a list of formatted chunks representing the differences in the text.

```APIDOC
## Text.diff()

### Description
Returns a list of formatted chunks that the current text corresponds to.

### Method
`diff`

### Response
#### Success Response
- **list[tuple[Any, dict[str, Any] | None]]** - A list of formatted chunks. Each item is a tuple containing the chunk's content and formatting attributes.

```

--------------------------------

### `items()`

Source: https://y-crdt.github.io/pycrdt/api_reference

Returns an iterable of key-value pairs for all root types currently defined in the document.

```APIDOC
## `items()`

### Returns

| Type  | Description  |
|---|---|
| `Iterable[tuple[str, T]]` | An iterable over the key-value pairs of document root types.  |

```

--------------------------------

### `get_state()`

Source: https://y-crdt.github.io/pycrdt/api_reference

Retrieves the current state of the document as a byte string, useful for snapshots and synchronization.

```APIDOC
## `get_state()`

### Returns

| Type  | Description  |
|---|---|
| `bytes` | The current document state.  |

```

--------------------------------

### Observe Text Changes

Source: https://y-crdt.github.io/pycrdt/usage

Register a callback function to handle changes in `Text` data types. The callback receives a `TextEvent` object containing details about the change.

```python
from pycrdt import TextEvent

def handle_changes(event: TextEvent):
    # process the event
    ...

text0_subscription_id = text0.observe(handle_changes)

```

--------------------------------

### Initialize Pycrdt Shared Data Types

Source: https://y-crdt.github.io/pycrdt/usage

Initialize Text, Array, and Map types with Python built-in counterparts. These are placeholders until inserted into a shared document.

```python
from pycrdt import Text, Array, Map

text0 = Text("Hello")
array0 = Array([0, "foo"])
map0 = Map({"key0": "value0"})

```

--------------------------------

### `handle_sync_message(message, ydoc)`

Source: https://y-crdt.github.io/pycrdt/api_reference

Processes a synchronization message on a document. It can return a SYNC_STEP2 reply if the input message was a SYNC_STEP1.

```APIDOC
## `handle_sync_message(message, ydoc)`

### Description
Processes a synchronization message on a document. Returns a SYNC_STEP2 reply message if the input message was a SYNC_STEP1.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
- **message** (bytes) - Required - A synchronization message.
- **ydoc** (Doc) - Required - The Doc that this message targets.

### Returns
- **bytes | None** - The SYNC_STEP2 reply message, if the message was a SYNC_STEP1.
```

--------------------------------

### values

Source: https://y-crdt.github.io/pycrdt/api_reference

Returns an iterable over the root types of the document.

```APIDOC
## `values()`

### Description
Returns an iterable over the document root types.

### Returns
- **Iterable[T]**: An iterable over the document root types.
```

--------------------------------

### Observe Deep Events

Source: https://y-crdt.github.io/pycrdt/api_reference

Subscribe a callback function to receive notifications for all events emitted by this XmlFragment and any nested collaborative types.

```python
observe_deep(callback)
```

--------------------------------

### TypedDoc and TypedMap for Key-Specific Types

Source: https://y-crdt.github.io/pycrdt/usage

Use TypedDoc and TypedMap to associate specific types with keys in your document and maps. This provides strong typing for nested structures and helps catch type errors during development.

```python
from pycrdt import Array, Text, TypedDoc, TypedMap

class MyMap(TypedMap):
    name: str
    toggle: bool
    nested: Array[bool]

class MyDoc(TypedDoc):
    map0: MyMap
    array0: Array[int]
    text0: Text

doc = MyDoc()

doc.map0.name = "foo"
doc.map0.toggle = False
doc.map0.toggle = 3  # error: Incompatible types in assignment (expression has type "int", variable has type "bool")  [assignment]
doc.array0 = Array([1, 2, 3])
doc.map0.nested = Array([4])  # error: List item 0 has incompatible type "int"; expected "bool"  [list-item]
doc.map0.nested = Array([False, True])
v0: str = doc.map0.name
v1: str = doc.map0.toggle  # error: Incompatible types in assignment (expression has type "bool", variable has type "str")  [assignment]
v2: bool = doc.map0.toggle
doc.map0.wrong_key0  # error: "MyMap" has no attribute "wrong_key0"  [attr-defined]
```

--------------------------------

### Encoder.write_var_uint(num)

Source: https://y-crdt.github.io/pycrdt/api_reference

Encodes an integer into the binary stream.

```APIDOC
## Encoder.write_var_uint(num)

### Description
Encodes a number into the binary stream.

### Parameters
#### Path Parameters
- **num** (int) - Required - The number to encode.
```

--------------------------------

### Array.__getitem__

Source: https://y-crdt.github.io/pycrdt/api_reference

Retrieves an item from the array at the specified index or a slice of items.

```APIDOC
## Array.__getitem__

### Description
Gets the item at the given index.

### Method
__getitem__

### Parameters
#### Parameters
- **key** (int | slice) - Required - The index or slice to retrieve.

### Returns
#### Returns
- **T | list[T]** - The item or list of items at the specified index/slice.

### Example
```python
Doc()["array"] = array = Array(["foo", "bar", "baz"])
assert array[1] == "bar"
```
```

--------------------------------

### Type-Annotated Document with Uniform Arrays

Source: https://y-crdt.github.io/pycrdt/usage

Declare a document where all root types are arrays of integers for static type analysis. Attempting to append incompatible types or retrieve incorrect types will result in errors.

```python
from pycrdt import Array, Doc

doc = Doc[Array[int]]()
array0 = doc.get("array0", type=Array[int])
array0.append(0)
array0.append("foo")  # error: Argument 1 to "append" of "Array" has incompatible type "str"; expected "int"
array1 = doc.get("array1", type=Array[str])  # error: Argument "type" to "get" of "Doc" has incompatible type "type[pycrdt._array.Array[Any]]"; expected "type[pycrdt._array.Array[int]]"
```

--------------------------------

### Send a message through a channel

Source: https://y-crdt.github.io/pycrdt/api_reference

Use the send() method to transmit a message over the channel.

```python
await channel.send(message)
```

--------------------------------

### insert

Source: https://y-crdt.github.io/pycrdt/api_reference

Inserts a string at a given index in the text, with optional attributes.

```APIDOC
## insert(index, value, attrs=None)

### Description
Inserts a string at a given index in the text, with optional attributes.

### Parameters
#### Path Parameters
- **index** (int) - Required - The index where to insert the string.
- **value** (str) - Required - The string to insert in the text.
- **attrs** (dict[str, Any] | None) - Optional - The formatting of attributes to apply.
```

--------------------------------

### Iterate over channel messages

Source: https://y-crdt.github.io/pycrdt/api_reference

Use an async for loop to receive messages from a channel until it is closed.

```python
async for message in channel:
    ...
```
Tokens:
Raw