43 lines
1.4 KiB
Python
43 lines
1.4 KiB
Python
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) |