Collaborative session implementation, 1

synctoy/snapshot.py

@@ -0,0 +1,43 @@
+from typing import Protocol
+
+from synctoy.data_model import NonNullRecord, ObjectId, RNull, Record
+
+
+class Selector(Protocol):
+    def includes_object_id(self, object_id: ObjectId):
+        raise NotImplementedError
+
+
+class SAll(object):
+    def includes_object_id(self, object_id: ObjectId):
+        return True
+
+class SList(object):
+    def __init__(self, object_ids: list[ObjectId]):
+        self._object_ids = object_ids
+
+    def includes_object_id(self, object_id: ObjectId):
+        return object_id in self._object_ids
+
+class Snapshot(object):
+    def __init__(self, selector: Selector, data: dict[ObjectId, NonNullRecord]):
+        self._selector = selector
+        self._data = data
+
+    def __getitem__(self, object_id: ObjectId):
+        if not self._selector.includes_object_id(object_id):
+            raise KeyError(f"{object_id} was not selected and cannot be viewed")
+
+        return self._data.get(object_id) or RNull(id=object_id)
+
+    def __setitem__(self, object_id: ObjectId, value: Record):
+        if not self._selector.includes_object_id(object_id):
+            raise KeyError(f"{object_id} was not selected and cannot be staged")
+
+        if isinstance(value, RNull):
+            self._data.pop(object_id, None)
+        else:
+            self._data[object_id] = value
+
+    def populated_records(self) -> dict[ObjectId, NonNullRecord]:
+        return dict(self._data)