0
# Position Management & Undo
1
2
## Overview
3
4
pycrdt provides robust position tracking and undo/redo functionality essential for collaborative editing. StickyIndex enables persistent position tracking that survives concurrent edits, while UndoManager provides comprehensive undo/redo operations with origin filtering and scope control. These features are crucial for building user-friendly collaborative editors.
5
6
## Position Management
7
8
### StickyIndex
9
10
Persistent position tracker that maintains its location during concurrent edits.
11
12
```python { .api }
13
class StickyIndex:
14
"""
15
Permanent position that maintains location during concurrent updates.
16
17
StickyIndex represents a position in a sequence (Text or Array) that
18
automatically adjusts when other clients make concurrent edits.
19
"""
20
21
@property
22
def assoc(self) -> Assoc:
23
"""Get the association type (before/after) for this position."""
24
25
def get_index(self, transaction: Transaction | None = None) -> int:
26
"""
27
Get the current index position.
28
29
Args:
30
transaction (Transaction, optional): Transaction context for reading position
31
32
Returns:
33
int: Current index position
34
"""
35
36
def encode(self) -> bytes:
37
"""
38
Encode the sticky index to binary format.
39
40
Returns:
41
bytes: Encoded position data
42
"""
43
44
def to_json(self) -> dict:
45
"""
46
Convert sticky index to JSON-serializable format.
47
48
Returns:
49
dict: JSON representation of the position
50
"""
51
52
@classmethod
53
def new(cls, sequence: Sequence, index: int, assoc: Assoc) -> Self:
54
"""
55
Create a new sticky index for a sequence.
56
57
Args:
58
sequence (Sequence): Text or Array to create position for
59
index (int): Initial position index
60
assoc (Assoc): Association type (BEFORE or AFTER)
61
62
Returns:
63
StickyIndex: New sticky index instance
64
"""
65
66
@classmethod
67
def decode(cls, data: bytes, sequence: Sequence | None = None) -> Self:
68
"""
69
Decode a sticky index from binary data.
70
71
Args:
72
data (bytes): Encoded position data
73
sequence (Sequence, optional): Sequence to attach position to
74
75
Returns:
76
StickyIndex: Decoded sticky index
77
"""
78
79
@classmethod
80
def from_json(cls, data: dict, sequence: Sequence | None = None) -> Self:
81
"""
82
Create sticky index from JSON data.
83
84
Args:
85
data (dict): JSON representation of position
86
sequence (Sequence, optional): Sequence to attach position to
87
88
Returns:
89
StickyIndex: Sticky index from JSON data
90
"""
91
```
92
93
### Assoc
94
95
Association type for sticky positions.
96
97
```python { .api }
98
class Assoc(IntEnum):
99
"""Specifies whether sticky index associates before or after position."""
100
101
AFTER = 0 # Associate with item after the position
102
BEFORE = -1 # Associate with item before the position
103
```
104
105
## Undo Management
106
107
### UndoManager
108
109
Comprehensive undo/redo operations with origin filtering and scope control.
110
111
```python { .api }
112
class UndoManager:
113
def __init__(
114
self,
115
*,
116
doc: Doc | None = None,
117
scopes: list[BaseType] = [],
118
capture_timeout_millis: int = 500,
119
timestamp: Callable[[], int] = timestamp,
120
) -> None:
121
"""
122
Create an undo manager for document operations.
123
124
Args:
125
doc (Doc, optional): Document to track changes for
126
scopes (list[BaseType]): List of shared types to track
127
capture_timeout_millis (int): Timeout for capturing operations into single undo step
128
timestamp (Callable): Function to generate timestamps
129
"""
130
131
@property
132
def undo_stack(self) -> list[StackItem]:
133
"""Get the list of undoable operations."""
134
135
@property
136
def redo_stack(self) -> list[StackItem]:
137
"""Get the list of redoable operations."""
138
139
def expand_scope(self, scope: BaseType) -> None:
140
"""
141
Add a shared type to the undo manager's scope.
142
143
Args:
144
scope (BaseType): Shared type to start tracking
145
"""
146
147
def include_origin(self, origin: Any) -> None:
148
"""
149
Include operations with specific origin in undo tracking.
150
151
Args:
152
origin: Origin identifier to include
153
"""
154
155
def exclude_origin(self, origin: Any) -> None:
156
"""
157
Exclude operations with specific origin from undo tracking.
158
159
Args:
160
origin: Origin identifier to exclude
161
"""
162
163
def can_undo(self) -> bool:
164
"""
165
Check if there are operations that can be undone.
166
167
Returns:
168
bool: True if undo is possible
169
"""
170
171
def undo(self) -> bool:
172
"""
173
Undo the last operation.
174
175
Returns:
176
bool: True if undo was performed
177
"""
178
179
def can_redo(self) -> bool:
180
"""
181
Check if there are operations that can be redone.
182
183
Returns:
184
bool: True if redo is possible
185
"""
186
187
def redo(self) -> bool:
188
"""
189
Redo the last undone operation.
190
191
Returns:
192
bool: True if redo was performed
193
"""
194
195
def clear(self) -> None:
196
"""Clear all undo and redo history."""
197
```
198
199
### StackItem
200
201
Undo stack item containing reversible operations.
202
203
```python { .api }
204
class StackItem:
205
"""
206
Represents a single undoable operation or group of operations.
207
208
StackItems are created automatically by the UndoManager and contain
209
the information needed to reverse document changes.
210
"""
211
```
212
213
## Usage Examples
214
215
### Basic Position Tracking
216
217
```python
218
from pycrdt import Doc, Text, StickyIndex, Assoc
219
220
doc = Doc()
221
text = doc.get("content", type=Text)
222
223
# Create initial content
224
text.insert(0, "Hello, world! This is a test document.")
225
226
# Create sticky positions
227
start_pos = text.sticky_index(7, Assoc.BEFORE) # Before "world"
228
end_pos = text.sticky_index(12, Assoc.AFTER) # After "world"
229
cursor_pos = text.sticky_index(20, Assoc.AFTER) # After "This"
230
231
print(f"Initial positions: start={start_pos.get_index()}, end={end_pos.get_index()}, cursor={cursor_pos.get_index()}")
232
233
# Make edits that affect positions
234
text.insert(0, "Well, ") # Insert at beginning
235
print(f"After insert at 0: start={start_pos.get_index()}, end={end_pos.get_index()}, cursor={cursor_pos.get_index()}")
236
237
text.insert(15, "beautiful ") # Insert within tracked region
238
print(f"After insert at 15: start={start_pos.get_index()}, end={end_pos.get_index()}, cursor={cursor_pos.get_index()}")
239
240
# Delete text before tracked positions
241
del text[0:6] # Remove "Well, "
242
print(f"After delete: start={start_pos.get_index()}, end={end_pos.get_index()}, cursor={cursor_pos.get_index()}")
243
244
# Get text at tracked positions
245
with doc.transaction() as txn:
246
start_idx = start_pos.get_index(txn)
247
end_idx = end_pos.get_index(txn)
248
tracked_text = text[start_idx:end_idx]
249
print(f"Tracked text: '{tracked_text}'")
250
```
251
252
### Position Persistence
253
254
```python
255
import json
256
from pycrdt import Doc, Text, StickyIndex, Assoc
257
258
doc = Doc()
259
text = doc.get("content", type=Text)
260
text.insert(0, "Persistent position tracking example.")
261
262
# Create positions
263
bookmark = text.sticky_index(10, Assoc.AFTER)
264
selection_start = text.sticky_index(15, Assoc.BEFORE)
265
selection_end = text.sticky_index(23, Assoc.AFTER)
266
267
# Serialize positions
268
bookmark_data = bookmark.to_json()
269
selection_data = {
270
"start": selection_start.to_json(),
271
"end": selection_end.to_json()
272
}
273
274
positions_json = json.dumps({
275
"bookmark": bookmark_data,
276
"selection": selection_data
277
}, indent=2)
278
279
print(f"Serialized positions:\n{positions_json}")
280
281
# Make changes to document
282
text.insert(5, "NEW ")
283
text.insert(30, " UPDATED")
284
285
# Deserialize positions in new session
286
doc2 = Doc()
287
text2 = doc2.get("content", type=Text)
288
289
# Apply the changes to new document
290
update = doc.get_update()
291
doc2.apply_update(update)
292
293
# Restore positions
294
positions_data = json.loads(positions_json)
295
restored_bookmark = StickyIndex.from_json(positions_data["bookmark"], text2)
296
restored_start = StickyIndex.from_json(positions_data["selection"]["start"], text2)
297
restored_end = StickyIndex.from_json(positions_data["selection"]["end"], text2)
298
299
print(f"Restored bookmark: {restored_bookmark.get_index()}")
300
print(f"Restored selection: {restored_start.get_index()}-{restored_end.get_index()}")
301
302
# Verify positions still track correctly
303
with doc2.transaction() as txn:
304
start_idx = restored_start.get_index(txn)
305
end_idx = restored_end.get_index(txn)
306
selected_text = text2[start_idx:end_idx]
307
print(f"Selected text: '{selected_text}'")
308
```
309
310
### Basic Undo/Redo Operations
311
312
```python
313
from pycrdt import Doc, Text, Array, UndoManager
314
315
doc = Doc()
316
text = doc.get("content", type=Text)
317
array = doc.get("items", type=Array)
318
319
# Create undo manager
320
undo_manager = UndoManager(doc=doc, scopes=[text, array])
321
322
# Make some changes
323
with doc.transaction(origin="user"):
324
text.insert(0, "Hello, world!")
325
326
print(f"Text: {str(text)}")
327
print(f"Can undo: {undo_manager.can_undo()}")
328
329
# Make more changes
330
with doc.transaction(origin="user"):
331
array.extend(["item1", "item2", "item3"])
332
333
print(f"Array: {list(array)}")
334
335
# Undo last operation
336
if undo_manager.can_undo():
337
undo_manager.undo()
338
print(f"After undo - Array: {list(array)}")
339
print(f"Can redo: {undo_manager.can_redo()}")
340
341
# Undo text changes
342
if undo_manager.can_undo():
343
undo_manager.undo()
344
print(f"After undo - Text: '{str(text)}'")
345
346
# Redo operations
347
if undo_manager.can_redo():
348
undo_manager.redo()
349
print(f"After redo - Text: '{str(text)}'")
350
351
if undo_manager.can_redo():
352
undo_manager.redo()
353
print(f"After redo - Array: {list(array)}")
354
```
355
356
### Origin Filtering
357
358
```python
359
from pycrdt import Doc, Text, UndoManager
360
361
doc = Doc()
362
text = doc.get("content", type=Text)
363
364
# Create undo manager that only tracks user operations
365
undo_manager = UndoManager(doc=doc, scopes=[text])
366
undo_manager.include_origin("user") # Only track "user" origin
367
368
# Make changes with different origins
369
with doc.transaction(origin="user"):
370
text.insert(0, "User change 1")
371
372
with doc.transaction(origin="system"):
373
text.insert(0, "System change - ") # This won't be tracked
374
375
with doc.transaction(origin="user"):
376
text.insert(len(text), " - User change 2")
377
378
print(f"Final text: {str(text)}")
379
print(f"Undo stack size: {len(undo_manager.undo_stack)}")
380
381
# Undo - should only undo user changes
382
undo_manager.undo()
383
print(f"After first undo: {str(text)}")
384
385
undo_manager.undo()
386
print(f"After second undo: {str(text)}")
387
388
# System change remains because it wasn't tracked
389
print(f"System change still present: {'System change' in str(text)}")
390
```
391
392
### Collaborative Undo with Position Tracking
393
394
```python
395
from pycrdt import Doc, Text, UndoManager, StickyIndex, Assoc
396
397
# Simulate collaborative editing with undo
398
doc1 = Doc(client_id=1)
399
doc2 = Doc(client_id=2)
400
401
text1 = doc1.get("document", type=Text)
402
text2 = doc2.get("document", type=Text)
403
404
# Create undo managers for each client
405
undo1 = UndoManager(doc=doc1, scopes=[text1])
406
undo1.include_origin("client1")
407
408
undo2 = UndoManager(doc=doc2, scopes=[text2])
409
undo2.include_origin("client2")
410
411
# Create position trackers
412
cursor1 = None
413
cursor2 = None
414
415
# Client 1 makes initial changes
416
with doc1.transaction(origin="client1"):
417
text1.insert(0, "Collaborative document. ")
418
cursor1 = text1.sticky_index(len(text1), Assoc.AFTER)
419
420
# Sync to client 2
421
update = doc1.get_update()
422
doc2.apply_update(update)
423
cursor2 = text2.sticky_index(10, Assoc.AFTER) # Position at "document"
424
425
print(f"Initial state: '{str(text1)}'")
426
427
# Client 2 makes concurrent changes
428
with doc2.transaction(origin="client2"):
429
pos = cursor2.get_index()
430
text2.insert(pos, " EDITED")
431
432
# Client 1 continues editing
433
with doc1.transaction(origin="client1"):
434
pos = cursor1.get_index()
435
text1.insert(pos, "More content from client 1.")
436
437
# Sync changes
438
update1 = doc1.get_update(doc2.get_state())
439
update2 = doc2.get_update(doc1.get_state())
440
441
doc2.apply_update(update1)
442
doc1.apply_update(update2)
443
444
print(f"After collaboration: '{str(text1)}'")
445
print(f"Client 2 sees: '{str(text2)}'")
446
447
# Each client can undo their own changes
448
print(f"Client 1 can undo: {undo1.can_undo()}")
449
print(f"Client 2 can undo: {undo2.can_undo()}")
450
451
# Client 1 undoes their changes
452
if undo1.can_undo():
453
undo1.undo() # Undo "More content from client 1."
454
print(f"Client 1 after undo: '{str(text1)}'")
455
456
if undo1.can_undo():
457
undo1.undo() # Undo initial text
458
print(f"Client 1 after second undo: '{str(text1)}'")
459
460
# Client 2's changes remain
461
print(f"Client 2's changes still present: {'EDITED' in str(text1)}")
462
```
463
464
### Advanced Undo Manager Configuration
465
466
```python
467
from pycrdt import Doc, Text, Array, Map, UndoManager
468
import time
469
470
def custom_timestamp():
471
"""Custom timestamp function."""
472
return int(time.time() * 1000)
473
474
doc = Doc()
475
text = doc.get("text", type=Text)
476
array = doc.get("array", type=Array)
477
map_data = doc.get("map", type=Map)
478
479
# Create undo manager with custom configuration
480
undo_manager = UndoManager(
481
doc=doc,
482
scopes=[text, array], # Only track text and array, not map
483
capture_timeout_millis=1000, # Group operations within 1 second
484
timestamp=custom_timestamp
485
)
486
487
# Configure origin filtering
488
undo_manager.include_origin("user")
489
undo_manager.exclude_origin("auto-save")
490
491
print("Making grouped changes (within timeout)...")
492
493
# Make changes quickly (will be grouped)
494
start_time = time.time()
495
with doc.transaction(origin="user"):
496
text.insert(0, "Quick ")
497
498
with doc.transaction(origin="user"):
499
text.insert(6, "changes ")
500
501
with doc.transaction(origin="user"):
502
array.append("item1")
503
504
elapsed = (time.time() - start_time) * 1000
505
print(f"Changes made in {elapsed:.1f}ms")
506
507
# Make changes to untracked type
508
with doc.transaction(origin="user"):
509
map_data["key"] = "value" # Not tracked
510
511
print(f"Undo stack size: {len(undo_manager.undo_stack)}")
512
513
# Wait for timeout then make another change
514
time.sleep(1.1) # Exceed capture timeout
515
516
with doc.transaction(origin="user"):
517
text.insert(len(text), "separate")
518
519
print(f"Undo stack size after timeout: {len(undo_manager.undo_stack)}")
520
521
# Make auto-save change (will be ignored)
522
with doc.transaction(origin="auto-save"):
523
text.insert(0, "[SAVED] ")
524
525
print(f"Text after auto-save: '{str(text)}'")
526
print(f"Undo stack size (auto-save ignored): {len(undo_manager.undo_stack)}")
527
528
# Undo operations
529
print("\nUndoing operations:")
530
while undo_manager.can_undo():
531
undo_manager.undo()
532
print(f"Text: '{str(text)}', Array: {list(array)}")
533
534
# Auto-save prefix remains (wasn't tracked)
535
print(f"Auto-save change remains: {str(text).startswith('[SAVED]')}")
536
```
537
538
### Undo Manager Scope Management
539
540
```python
541
from pycrdt import Doc, Text, Array, Map, UndoManager
542
543
doc = Doc()
544
text = doc.get("text", type=Text)
545
array = doc.get("array", type=Array)
546
map_data = doc.get("map", type=Map)
547
548
# Create undo manager with initial scope
549
undo_manager = UndoManager(doc=doc, scopes=[text])
550
551
# Make changes to tracked type
552
with doc.transaction(origin="user"):
553
text.insert(0, "Tracked text")
554
555
# Make changes to untracked type
556
with doc.transaction(origin="user"):
557
array.append("untracked item")
558
559
print(f"Initial undo stack size: {len(undo_manager.undo_stack)}")
560
561
# Expand scope to include array
562
undo_manager.expand_scope(array)
563
564
# Now array changes will be tracked
565
with doc.transaction(origin="user"):
566
array.append("tracked item")
567
text.insert(0, "More ")
568
569
print(f"After scope expansion: {len(undo_manager.undo_stack)}")
570
571
# Add map to scope
572
undo_manager.expand_scope(map_data)
573
574
with doc.transaction(origin="user"):
575
map_data["key1"] = "value1"
576
map_data["key2"] = "value2"
577
578
print(f"After adding map: {len(undo_manager.undo_stack)}")
579
580
# Undo all tracked operations
581
print("\nUndoing all operations:")
582
while undo_manager.can_undo():
583
print(f"Before undo: text='{str(text)}', array={list(array)}, map={dict(map_data.items())}")
584
undo_manager.undo()
585
586
# The first array item remains (was added before array was in scope)
587
print(f"Final state: text='{str(text)}', array={list(array)}, map={dict(map_data.items())}")
588
```
589
590
### Position-Aware Undo Operations
591
592
```python
593
from pycrdt import Doc, Text, UndoManager, StickyIndex, Assoc
594
595
class PositionAwareEditor:
596
"""Editor that maintains cursor position through undo/redo."""
597
598
def __init__(self, doc: Doc):
599
self.doc = doc
600
self.text = doc.get("content", type=Text)
601
self.undo_manager = UndoManager(doc=doc, scopes=[self.text])
602
self.cursor_pos = self.text.sticky_index(0, Assoc.AFTER)
603
604
def insert_text(self, text: str, origin="user"):
605
"""Insert text at cursor position."""
606
with self.doc.transaction(origin=origin):
607
pos = self.cursor_pos.get_index()
608
self.text.insert(pos, text)
609
# Update cursor to end of inserted text
610
self.cursor_pos = self.text.sticky_index(pos + len(text), Assoc.AFTER)
611
612
def delete_range(self, start: int, end: int, origin="user"):
613
"""Delete text range and update cursor."""
614
with self.doc.transaction(origin=origin):
615
del self.text[start:end]
616
# Move cursor to start of deleted range
617
self.cursor_pos = self.text.sticky_index(start, Assoc.AFTER)
618
619
def move_cursor(self, position: int):
620
"""Move cursor to specific position."""
621
position = max(0, min(position, len(self.text)))
622
self.cursor_pos = self.text.sticky_index(position, Assoc.AFTER)
623
624
def get_cursor_position(self) -> int:
625
"""Get current cursor position."""
626
return self.cursor_pos.get_index()
627
628
def undo(self):
629
"""Undo with cursor position awareness."""
630
if self.undo_manager.can_undo():
631
old_pos = self.get_cursor_position()
632
self.undo_manager.undo()
633
# Try to maintain reasonable cursor position
634
new_pos = min(old_pos, len(self.text))
635
self.move_cursor(new_pos)
636
return True
637
return False
638
639
def redo(self):
640
"""Redo with cursor position awareness."""
641
if self.undo_manager.can_redo():
642
old_pos = self.get_cursor_position()
643
self.undo_manager.redo()
644
# Try to maintain reasonable cursor position
645
new_pos = min(old_pos, len(self.text))
646
self.move_cursor(new_pos)
647
return True
648
return False
649
650
# Example usage
651
doc = Doc()
652
editor = PositionAwareEditor(doc)
653
654
print("Position-aware editing demo:")
655
656
# Type some text
657
editor.insert_text("Hello, ")
658
print(f"Text: '{str(editor.text)}', Cursor: {editor.get_cursor_position()}")
659
660
editor.insert_text("world!")
661
print(f"Text: '{str(editor.text)}', Cursor: {editor.get_cursor_position()}")
662
663
# Move cursor and insert
664
editor.move_cursor(7) # Between "Hello, " and "world!"
665
editor.insert_text("beautiful ")
666
print(f"Text: '{str(editor.text)}', Cursor: {editor.get_cursor_position()}")
667
668
# Undo operations
669
print("\nUndoing operations:")
670
while editor.undo():
671
print(f"Text: '{str(editor.text)}', Cursor: {editor.get_cursor_position()}")
672
673
# Redo operations
674
print("\nRedoing operations:")
675
while editor.redo():
676
print(f"Text: '{str(editor.text)}', Cursor: {editor.get_cursor_position()}")
677
```
678
679
## Error Handling
680
681
```python
682
from pycrdt import Doc, Text, StickyIndex, UndoManager, Assoc
683
684
doc = Doc()
685
text = doc.get("content", type=Text)
686
687
try:
688
# Invalid sticky index operations
689
invalid_pos = text.sticky_index(-1, Assoc.AFTER) # May raise ValueError
690
691
except ValueError as e:
692
print(f"StickyIndex error: {e}")
693
694
try:
695
# Invalid undo manager operations
696
undo_manager = UndoManager(doc=doc, scopes=[text])
697
698
# Try to undo when nothing to undo
699
if not undo_manager.can_undo():
700
result = undo_manager.undo() # Returns False, doesn't raise
701
print(f"Undo result when nothing to undo: {result}")
702
703
# Invalid scope expansion
704
undo_manager.expand_scope(None) # May raise TypeError
705
706
except (TypeError, ValueError) as e:
707
print(f"UndoManager error: {e}")
708
709
try:
710
# Position encoding/decoding errors
711
pos = text.sticky_index(0, Assoc.AFTER)
712
invalid_data = b"invalid"
713
714
StickyIndex.decode(invalid_data, text) # May raise decoding error
715
716
except Exception as e:
717
print(f"Position decoding error: {e}")
718
```