0
# Map Operations
1

2
## Overview
3

4
The `Map` type in pycrdt provides collaborative dictionary/map functionality with automatic conflict resolution across multiple clients. It supports a complete dict-like interface with additional collaborative features like change tracking, deep observation, and type-safe variants.
5

6
## Core Types
7

8
### Map
9

10
Collaborative map with dict-like interface and change tracking.
11

12
```python { .api }
13
class Map[T]:
14
    def __init__(
15
        self,
16
        init: dict[str, T] | None = None,
17
        *,
18
        _doc: Doc | None = None,
19
        _integrated: _Map | None = None,
20
    ) -> None:
21
        """
22
        Create a new collaborative map.
23
        
24
        Args:
25
            init (dict, optional): Initial map contents
26
            _doc (Doc, optional): Parent document
27
            _integrated (_Map, optional): Native map instance
28
        """
29

30
    # Dict-like interface
31
    def __len__(self) -> int:
32
        """Get the number of key-value pairs in the map."""
33

34
    def __str__(self) -> str:
35
        """Get string representation of the map."""
36

37
    def __iter__(self) -> Iterable[str]:
38
        """Iterate over map keys."""
39

40
    def __contains__(self, item: str) -> bool:
41
        """Check if key exists in map."""
42

43
    def __getitem__(self, key: str) -> T:
44
        """Get value by key."""
45

46
    def __setitem__(self, key: str, value: T) -> None:
47
        """Set value for key."""
48

49
    def __delitem__(self, key: str) -> None:
50
        """Delete key-value pair."""
51

52
    # Map manipulation methods
53
    def get(self, key: str, default_value: T_DefaultValue = None) -> T | T_DefaultValue | None:
54
        """
55
        Get value for key with optional default.
56
        
57
        Args:
58
            key (str): Key to lookup
59
            default_value: Default value if key not found
60
            
61
        Returns:
62
            T | T_DefaultValue | None: Value or default
63
        """
64

65
    def pop(self, key: str, default_value: T_DefaultValue = None) -> T | T_DefaultValue:
66
        """
67
        Remove key and return its value.
68
        
69
        Args:
70
            key (str): Key to remove
71
            default_value: Default value if key not found
72
            
73
        Returns:
74
            T | T_DefaultValue: Removed value or default
75
        """
76

77
    def keys(self) -> Iterable[str]:
78
        """Get all keys in the map."""
79

80
    def values(self) -> Iterable[T]:
81
        """Get all values in the map."""
82

83
    def items(self) -> Iterable[tuple[str, T]]:
84
        """Get all key-value pairs as tuples."""
85

86
    def clear(self) -> None:
87
        """Remove all key-value pairs from the map."""
88

89
    def update(self, value: dict[str, T]) -> None:
90
        """
91
        Update map with key-value pairs from another dict.
92
        
93
        Args:
94
            value (dict): Dictionary to merge into this map
95
        """
96

97
    def to_py(self) -> dict[str, T] | None:
98
        """
99
        Convert map to a Python dictionary.
100
        
101
        Returns:
102
            dict | None: Map contents as dict, or None if empty
103
        """
104

105
    def observe(self, callback: Callable[[MapEvent], None]) -> Subscription:
106
        """
107
        Observe map changes.
108
        
109
        Args:
110
            callback: Function called when map changes occur
111
            
112
        Returns:
113
            Subscription: Handle for unsubscribing
114
        """
115

116
    def observe_deep(self, callback: Callable[[list[MapEvent]], None]) -> Subscription:
117
        """
118
        Observe deep changes including nested structures.
119
        
120
        Args:
121
            callback: Function called with list of change events
122
            
123
        Returns:
124
            Subscription: Handle for unsubscribing
125
        """
126

127
    def unobserve(self, subscription: Subscription) -> None:
128
        """
129
        Remove an event observer.
130
        
131
        Args:
132
            subscription: Subscription handle to remove
133
        """
134

135
    async def events(
136
        self, 
137
        deep: bool = False, 
138
        max_buffer_size: float = float("inf")
139
    ) -> MemoryObjectReceiveStream:
140
        """
141
        Get an async stream of map events.
142
        
143
        Args:
144
            deep (bool): Include deep change events
145
            max_buffer_size (float): Maximum event buffer size
146
            
147
        Returns:
148
            MemoryObjectReceiveStream: Async event stream
149
        """
150
```
151

152
### MapEvent
153

154
Event emitted when map changes occur.
155

156
```python { .api }
157
class MapEvent:
158
    @property
159
    def target(self) -> Map:
160
        """Get the map that changed."""
161

162
    @property
163
    def keys(self) -> list[str]:
164
        """Get the list of keys that changed."""
165

166
    @property
167
    def path(self) -> list[int | str]:
168
        """Get the path to the changed map within the document structure."""
169
```
170

171
### TypedMap
172

173
Type-safe wrapper for Map with typed attributes.
174

175
```python { .api }
176
class TypedMap:
177
    """
178
    Type-safe map container with runtime type checking.
179
    
180
    Usage:
181
        class UserMap(TypedMap):
182
            name: str
183
            age: int
184
            active: bool
185
        
186
        user = UserMap()
187
        user.name = "Alice"    # Type-safe
188
        user.age = 30          # Type-safe
189
        name: str = user.name  # Typed access
190
    """
191
```
192

193
## Usage Examples
194

195
### Basic Map Operations
196

197
```python
198
from pycrdt import Doc, Map
199

200
doc = Doc()
201
config = doc.get("config", type=Map)
202

203
# Basic dict operations
204
config["theme"] = "dark"
205
config["font_size"] = 14
206
config["auto_save"] = True
207
print(len(config))  # 3
208

209
# Dict-like access
210
print(config["theme"])           # "dark"
211
print(config.get("missing"))     # None
212
print(config.get("missing", "default"))  # "default"
213

214
# Check keys
215
print("theme" in config)         # True
216
print("color" in config)         # False
217

218
# Iteration
219
for key in config:
220
    print(f"{key}: {config[key]}")
221

222
for key, value in config.items():
223
    print(f"{key} = {value}")
224
```
225

226
### Map Manipulation
227

228
```python
229
from pycrdt import Doc, Map
230

231
doc = Doc()
232
settings = doc.get("settings", type=Map)
233

234
# Build settings
235
settings.update({
236
    "width": 800,
237
    "height": 600,
238
    "fullscreen": False,
239
    "vsync": True
240
})
241

242
# Modify individual settings
243
settings["width"] = 1024
244
settings["height"] = 768
245

246
# Remove settings
247
old_value = settings.pop("vsync")
248
print(f"Removed vsync: {old_value}")
249

250
# Remove with default
251
fps_limit = settings.pop("fps_limit", 60)
252
print(f"FPS limit: {fps_limit}")
253

254
# Delete by key
255
del settings["fullscreen"]
256

257
# Check current state
258
print(f"Current settings: {dict(settings.items())}")
259
```
260

261
### Nested Data Structures
262

263
```python
264
from pycrdt import Doc, Map, Array
265

266
doc = Doc()
267
user_profile = doc.get("profile", type=Map)
268

269
# Create nested structure
270
user_profile["personal"] = Map()
271
user_profile["personal"]["name"] = "Alice"
272
user_profile["personal"]["email"] = "alice@example.com"
273

274
user_profile["preferences"] = Map()
275
user_profile["preferences"]["theme"] = "dark"
276
user_profile["preferences"]["notifications"] = True
277

278
user_profile["tags"] = Array()
279
user_profile["tags"].extend(["developer", "python", "collaborative"])
280

281
# Access nested data
282
print(user_profile["personal"]["name"])        # "Alice"
283
print(user_profile["preferences"]["theme"])    # "dark"
284
print(list(user_profile["tags"]))             # ["developer", "python", "collaborative"]
285

286
# Modify nested structures
287
user_profile["personal"]["age"] = 30
288
user_profile["tags"].append("crdt")
289
```
290

291
### Type-Safe Maps
292

293
```python
294
from pycrdt import TypedMap, Doc
295

296
class UserProfile(TypedMap):
297
    name: str
298
    age: int
299
    email: str
300
    active: bool
301

302
class DatabaseConfig(TypedMap):
303
    host: str
304
    port: int
305
    ssl_enabled: bool
306
    timeout: float
307

308
doc = Doc()
309

310
# Create typed maps
311
user = UserProfile()
312
db_config = DatabaseConfig()
313

314
# Type-safe operations
315
user.name = "Bob"           # OK
316
user.age = 25               # OK
317
user.email = "bob@test.com" # OK
318
user.active = True          # OK
319

320
db_config.host = "localhost"
321
db_config.port = 5432
322
db_config.ssl_enabled = True
323
db_config.timeout = 30.0
324

325
try:
326
    user.age = "not a number"  # May raise TypeError
327
except TypeError as e:
328
    print(f"Type error: {e}")
329

330
# Typed access
331
name: str = user.name       # Typed
332
port: int = db_config.port  # Typed
333
```
334

335
### Event Observation
336

337
```python
338
from pycrdt import Doc, Map, MapEvent
339

340
doc = Doc()
341
data = doc.get("data", type=Map)
342

343
def on_map_change(event: MapEvent):
344
    print(f"Map changed: {event.target}")
345
    print(f"Changed keys: {event.keys}")
346
    print(f"Path: {event.path}")
347

348
# Subscribe to changes
349
subscription = data.observe(on_map_change)
350

351
# Make changes to trigger events
352
data["key1"] = "value1"
353
data["key2"] = "value2"
354
data.update({"key3": "value3", "key4": "value4"})
355
del data["key1"]
356

357
# Clean up
358
data.unobserve(subscription)
359
```
360

361
### Deep Event Observation
362

363
```python
364
from pycrdt import Doc, Map, Array
365

366
doc = Doc()
367
root = doc.get("root", type=Map)
368

369
# Create nested structure
370
root["level1"] = Map()
371
root["level1"]["level2"] = Map()
372
root["level1"]["level2"]["data"] = Array()
373

374
def on_deep_change(events):
375
    print(f"Deep changes detected: {len(events)} events")
376
    for event in events:
377
        if hasattr(event, 'keys'):  # MapEvent
378
            print(f"  Map change at {event.path}: keys {event.keys}")
379
        elif hasattr(event, 'delta'):  # ArrayEvent
380
            print(f"  Array change at {event.path}: {event.delta}")
381

382
# Subscribe to deep changes
383
subscription = root.observe_deep(on_deep_change)
384

385
# Make nested changes
386
root["level1"]["level2"]["data"].append("item1")
387
root["level1"]["level2"]["new_key"] = "new_value"
388
root["level1"]["another_map"] = Map()
389

390
# Clean up
391
root.unobserve(subscription)
392
```
393

394
### Async Event Streaming
395

396
```python
397
import anyio
398
from pycrdt import Doc, Map
399

400
async def monitor_map_changes(map_obj: Map):
401
    async with map_obj.events() as event_stream:
402
        async for event in event_stream:
403
            print(f"Map event: keys {event.keys}")
404

405
doc = Doc()
406
config = doc.get("config", type=Map)
407

408
async def main():
409
    async with anyio.create_task_group() as tg:
410
        tg.start_soon(monitor_map_changes, config)
411
        
412
        # Make changes
413
        await anyio.sleep(0.1)
414
        config["setting1"] = "value1"
415
        await anyio.sleep(0.1)
416
        config.update({"setting2": "value2", "setting3": "value3"})
417
        await anyio.sleep(0.1)
418

419
anyio.run(main)
420
```
421

422
### Collaborative Map Editing
423

424
```python
425
from pycrdt import Doc, Map
426

427
# Simulate two clients editing the same map
428
doc1 = Doc(client_id=1)
429
doc2 = Doc(client_id=2)
430

431
config1 = doc1.get("shared_config", type=Map)
432
config2 = doc2.get("shared_config", type=Map)
433

434
# Client 1 sets initial configuration
435
with doc1.transaction(origin="client1"):
436
    config1.update({
437
        "theme": "light",
438
        "font_size": 12,
439
        "auto_save": True
440
    })
441

442
# Sync to client 2
443
update = doc1.get_update()
444
doc2.apply_update(update)
445
print(f"Client 2 config: {dict(config2.items())}")
446

447
# Client 2 makes concurrent changes
448
with doc2.transaction(origin="client2"):
449
    config2["theme"] = "dark"          # Conflict with client 1
450
    config2["line_numbers"] = True     # New setting
451
    config2["font_size"] = 14          # Different value
452

453
# Both clients make more changes
454
with doc1.transaction(origin="client1"):
455
    config1["word_wrap"] = True        # New setting from client 1
456
    
457
with doc2.transaction(origin="client2"):
458
    config2["auto_save"] = False       # Change existing setting
459

460
# Sync changes
461
update1 = doc1.get_update(doc2.get_state())
462
update2 = doc2.get_update(doc1.get_state())
463

464
doc2.apply_update(update1)
465
doc1.apply_update(update2)
466

467
# Both clients now have consistent state
468
print(f"Client 1 final: {dict(config1.items())}")
469
print(f"Client 2 final: {dict(config2.items())}")
470
```
471

472
### Complex Data Management
473

474
```python
475
from pycrdt import Doc, Map, Array
476

477
doc = Doc()
478
database = doc.get("database", type=Map)
479

480
# Create complex data structure
481
database["users"] = Map()
482
database["posts"] = Map()
483
database["comments"] = Map()
484

485
# Add users
486
users = database["users"]
487
users["1"] = Map()
488
users["1"].update({"name": "Alice", "email": "alice@test.com", "posts": []})
489

490
users["2"] = Map()
491
users["2"].update({"name": "Bob", "email": "bob@test.com", "posts": []})
492

493
# Add posts
494
posts = database["posts"]
495
posts["1"] = Map()
496
posts["1"].update({
497
    "title": "First Post",
498
    "content": "Hello, world!",
499
    "author_id": "1",
500
    "comments": []
501
})
502

503
posts["2"] = Map()
504
posts["2"].update({
505
    "title": "Second Post", 
506
    "content": "More content",
507
    "author_id": "2",
508
    "comments": []
509
})
510

511
# Link posts to users
512
users["1"]["posts"].append("1")
513
users["2"]["posts"].append("2")
514

515
# Add comments
516
comments = database["comments"]
517
comments["1"] = Map()
518
comments["1"].update({
519
    "content": "Great post!",
520
    "author_id": "2",
521
    "post_id": "1"
522
})
523

524
posts["1"]["comments"].append("1")
525

526
# Query operations
527
def get_user_posts(database: Map, user_id: str) -> list:
528
    """Get all posts by a user."""
529
    user = database["users"][user_id]
530
    post_ids = user["posts"]
531
    posts_data = []
532
    
533
    for post_id in post_ids:
534
        post = database["posts"][post_id]
535
        posts_data.append({
536
            "id": post_id,
537
            "title": post["title"],
538
            "content": post["content"]
539
        })
540
    
541
    return posts_data
542

543
def get_post_with_comments(database: Map, post_id: str) -> dict:
544
    """Get post with all its comments."""
545
    post = database["posts"][post_id]
546
    comment_ids = post["comments"]
547
    
548
    comments_data = []
549
    for comment_id in comment_ids:
550
        comment = database["comments"][comment_id]
551
        author = database["users"][comment["author_id"]]
552
        comments_data.append({
553
            "content": comment["content"],
554
            "author": author["name"]
555
        })
556
    
557
    return {
558
        "title": post["title"],
559
        "content": post["content"],
560
        "comments": comments_data
561
    }
562

563
# Use query operations
564
alice_posts = get_user_posts(database, "1")
565
print(f"Alice's posts: {alice_posts}")
566

567
first_post = get_post_with_comments(database, "1")
568
print(f"First post with comments: {first_post}")
569
```
570

571
### Map Serialization and Persistence
572

573
```python
574
import json
575
from pycrdt import Doc, Map, Array
576

577
def serialize_map(map_obj: Map) -> dict:
578
    """Serialize a collaborative map to JSON-compatible dict."""
579
    result = {}
580
    for key, value in map_obj.items():
581
        if isinstance(value, Map):
582
            result[key] = serialize_map(value)
583
        elif isinstance(value, Array):
584
            result[key] = serialize_array(value)
585
        else:
586
            result[key] = value
587
    return result
588

589
def serialize_array(array_obj: Array) -> list:
590
    """Serialize a collaborative array to JSON-compatible list."""
591
    result = []
592
    for item in array_obj:
593
        if isinstance(item, Map):
594
            result.append(serialize_map(item))
595
        elif isinstance(item, Array):
596
            result.append(serialize_array(item))
597
        else:
598
            result.append(item)
599
    return result
600

601
def deserialize_to_map(data: dict, doc: Doc) -> Map:
602
    """Deserialize dict to collaborative map."""
603
    map_obj = Map()
604
    for key, value in data.items():
605
        if isinstance(value, dict):
606
            map_obj[key] = deserialize_to_map(value, doc)
607
        elif isinstance(value, list):
608
            map_obj[key] = deserialize_to_array(value, doc)
609
        else:
610
            map_obj[key] = value
611
    return map_obj
612

613
def deserialize_to_array(data: list, doc: Doc) -> Array:
614
    """Deserialize list to collaborative array."""
615
    array_obj = Array()
616
    for item in data:
617
        if isinstance(item, dict):
618
            array_obj.append(deserialize_to_map(item, doc))
619
        elif isinstance(item, list):
620
            array_obj.append(deserialize_to_array(item, doc))
621
        else:
622
            array_obj.append(item)
623
    return array_obj
624

625
# Example usage
626
doc = Doc()
627
config = doc.get("config", type=Map)
628

629
# Build complex configuration
630
config["database"] = Map()
631
config["database"]["host"] = "localhost"
632
config["database"]["port"] = 5432
633

634
config["features"] = Array()
635
config["features"].extend(["auth", "logging", "caching"])
636

637
# Serialize to JSON
638
config_dict = serialize_map(config)
639
json_str = json.dumps(config_dict, indent=2)
640
print(f"Serialized config:\n{json_str}")
641

642
# Deserialize back
643
loaded_data = json.loads(json_str)
644
new_doc = Doc()
645
restored_config = deserialize_to_map(loaded_data, new_doc)
646
print(f"Restored config: {dict(restored_config.items())}")
647
```
648

649
## Error Handling
650

651
```python
652
from pycrdt import Doc, Map
653

654
doc = Doc()
655
data = doc.get("data", type=Map)
656

657
try:
658
    # Key not found
659
    value = data["nonexistent"]  # May raise KeyError
660
    
661
    # Invalid operations
662
    del data["nonexistent"]      # May raise KeyError
663
    
664
    # Type mismatches in typed maps
665
    class StrictMap(TypedMap):
666
        number_field: int
667
    
668
    strict_map = StrictMap()
669
    strict_map.number_field = "string"  # May raise TypeError
670
    
671
except (KeyError, TypeError, ValueError) as e:
672
    print(f"Map operation failed: {e}")
673

674
# Safe operations
675
value = data.get("nonexistent", "default")  # Returns "default"
676
removed = data.pop("nonexistent", None)     # Returns None
677
```