forked from aosabook/500lines
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathlogical.py
More file actions
75 lines (59 loc) · 2.11 KB
/
Copy pathlogical.py
File metadata and controls
75 lines (59 loc) · 2.11 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
class ValueRef(object):
def prepare_to_store(self, storage):
pass
@staticmethod
def referent_to_string(referent):
return referent.encode('utf-8')
@staticmethod
def string_to_referent(string):
return string.decode('utf-8')
def __init__(self, referent=None, address=0):
self._referent = referent
self._address = address
@property
def address(self):
return self._address
def get(self, storage):
if self._referent is None and self._address:
self._referent = self.string_to_referent(storage.read(self._address))
return self._referent
def store(self, storage):
if self._referent is not None and not self._address:
self.prepare_to_store(storage)
self._address = storage.write(self.referent_to_string(self._referent))
class LogicalBase(object):
node_ref_class = None
value_ref_class = ValueRef
def __init__(self, storage):
self._storage = storage
self._refresh_tree_ref()
def commit(self):
self._tree_ref.store(self._storage)
self._storage.commit_root_address(self._tree_ref.address)
def _refresh_tree_ref(self):
self._tree_ref = self.node_ref_class(
address=self._storage.get_root_address())
def get(self, key):
if not self._storage.locked:
self._refresh_tree_ref()
return self._get(self._follow(self._tree_ref), key)
def set(self, key, value):
if self._storage.lock():
self._refresh_tree_ref()
self._tree_ref = self._insert(
self._follow(self._tree_ref), key, self.value_ref_class(value))
def pop(self, key):
if self._storage.lock():
self._refresh_tree_ref()
self._tree_ref = self._delete(
self._follow(self._tree_ref), key)
def _follow(self, ref):
return ref.get(self._storage)
def __len__(self):
if not self._storage.locked:
self._refresh_tree_ref()
root = self._follow(self._tree_ref)
if root:
return root.length
else:
return 0