commit 0eaf032b894ba985c49995e33a4d61cab5e8924e
parent d2bca5134248f49c2935bfda5390b44da69e0250
Author: lash <dev@holbrook.no>
Date: Mon, 31 Jan 2022 11:23:51 +0000
Add item lifetime controller interface
Diffstat:
5 files changed, 184 insertions(+), 8 deletions(-)
diff --git a/CHANGELOG b/CHANGELOG
@@ -1,3 +1,5 @@
+- 0.0.4
+ * Add item lifetime control
- 0.0.3
* Split match to 2-element tuple, returning complex value and simple values separately
- 0.0.2
diff --git a/setup.cfg b/setup.cfg
@@ -1,6 +1,6 @@
[metadata]
name = shep
-version = 0.0.3
+version = 0.0.4
description = Multi-state key value stores using bitmaskings
author = Louis Holbrook
author_email = dev@holbrook.no
diff --git a/shep/error.py b/shep/error.py
@@ -4,3 +4,15 @@ class StateExists(Exception):
class StateInvalid(Exception):
pass
+
+
+class StateItemExists(Exception):
+ pass
+
+
+class StateItemNotFound(Exception):
+ pass
+
+
+class StateCorruptionError(RuntimeError):
+ pass
diff --git a/shep/state.py b/shep/state.py
@@ -1,22 +1,23 @@
-# standard imports
-import enum
-
# local imports
from shep.error import (
StateExists,
StateInvalid,
+ StateItemExists,
+ StateItemNotFound,
)
class State:
- def __init__(self, bits, logger=None, store=None):
+ def __init__(self, bits, logger=None, store_factory=None):
self.__bits = bits
self.__limit = (1 << bits) - 1
self.__c = 0
self.__reverse = {}
- self.__logger = logger
- self.__store = store
+
+ self.NEW = 0
+ self.__items = {self.NEW: []}
+ self.__items_reverse = {}
def __is_pure(self, v):
@@ -28,9 +29,13 @@ class State:
return c == v
- def __check_name(self, k):
+ def __check_name_valid(self, k):
if not k.isalpha():
raise ValueError('only alpha')
+
+ def __check_name(self, k):
+ self.__check_name_valid(k)
+
k = k.upper()
try:
getattr(self, k)
@@ -67,6 +72,31 @@ class State:
self.__c += 1
+ def __check_item(self, item):
+ if self.__items_reverse.get(item) != None:
+ raise StateItemExists(item)
+
+
+ def __add_state_list(self, state, item):
+ if self.__items.get(state) == None:
+ self.__items[state] = []
+ self.__items[state].append(item)
+ self.__items_reverse[item] = state
+
+
+ def __state_list_index(self, item, state_list):
+ idx = -1
+ try:
+ idx = state_list.index(item)
+ except ValueError:
+ pass
+
+ if idx == -1:
+ raise StateCorruptionError() # should have state int here as value
+
+ return idx
+
+
def add(self, k):
v = 1 << self.__c
k = self.__check_name(k)
@@ -111,3 +141,56 @@ class State:
c <<= 1
return (alias, r,)
+
+
+ def put(self, item, state=None):
+ if state == None:
+ state = self.NEW
+ elif self.__reverse.get(state) == None:
+ raise StateInvalid(state)
+ self.__check_item(item)
+ self.__add_state_list(state, item)
+
+
+ def move(self, item, to_state):
+ current_state = self.__items_reverse.get(item)
+ if current_state == None:
+ raise StateItemNotFound(item)
+
+ new_state = self.__reverse.get(to_state)
+ if new_state == None:
+ raise StateInvalid(to_state)
+
+ current_state_list = self.__items.get(current_state)
+ if current_state_list == None:
+ raise StateCorruptionError(current_state)
+
+ idx = self.__state_list_index(item, current_state_list)
+
+ new_state_list = self.__items.get(to_state)
+ if current_state_list == None:
+ raise StateCorruptionError(to_state)
+
+ self.__add_state_list(to_state, item)
+ current_state_list.pop(idx)
+
+
+
+ def purge(self, item):
+ current_state = self.__items_reverse.get(item)
+ if current_state == None:
+ raise StateItemNotFound(item)
+ del self.__items_reverse[item]
+
+ current_state_list = self.__items.get(current_state)
+
+ idx = self.__state_list_index(item, current_state_list)
+
+ current_state_list.pop(idx)
+
+
+ def state(self, item):
+ state = self.__items_reverse.get(item)
+ if state == None:
+ raise StateItemNotFound(item)
+ return state
diff --git a/tests/test_item.py b/tests/test_item.py
@@ -0,0 +1,79 @@
+# standard imports
+import unittest
+
+# local imports
+from shep import State
+from shep.error import (
+ StateExists,
+ StateItemExists,
+ StateInvalid,
+ StateItemNotFound,
+ )
+
+
+class TestStateItems(unittest.TestCase):
+
+ def setUp(self):
+ self.states = State(4)
+ self.states.add('foo')
+ self.states.add('bar')
+ self.states.add('baz')
+ self.states.alias('xyzzy', self.states.BAZ | self.states.BAR)
+ self.states.alias('plugh', self.states.FOO | self.states.BAR)
+
+
+ def test_put(self):
+ item = b'foo'
+
+ # put in initial (no) state
+ self.states.put(item)
+
+ with self.assertRaises(StateItemExists):
+ self.states.put(item)
+
+ with self.assertRaises(StateItemExists):
+ self.states.put(item, self.states.BAZ)
+
+
+ def test_item_state(self):
+ item = b'foo'
+ self.states.put(item, self.states.XYZZY)
+ self.assertEqual(self.states.state(item), self.states.XYZZY)
+
+
+ def test_item_move(self):
+ item = b'foo'
+ self.states.put(item, self.states.FOO)
+ self.states.move(item, self.states.BAR)
+ self.assertEqual(self.states.state(item), self.states.BAR)
+
+
+ def test_item_move_from_alias(self):
+ item = b'foo'
+ self.states.put(item, self.states.FOO)
+ self.states.move(item, self.states.XYZZY)
+ self.assertEqual(self.states.state(item), self.states.XYZZY)
+ self.states.move(item, self.states.BAR)
+ self.assertEqual(self.states.state(item), self.states.BAR)
+
+
+ def test_item_move_from_new(self):
+ item = b'foo'
+ self.states.put(item)
+ self.assertEqual(self.states.state(item), self.states.NEW)
+ self.states.move(item, self.states.XYZZY)
+ self.assertEqual(self.states.state(item), self.states.XYZZY)
+
+
+ def test_item_purge(self):
+ item = b'foo'
+ self.states.put(item, self.states.BAZ)
+ self.assertEqual(self.states.state(item), self.states.BAZ)
+ self.states.purge(item)
+ with self.assertRaises(StateItemNotFound):
+ self.states.state(item)
+
+
+
+if __name__ == '__main__':
+ unittest.main()
