Source code for sbws.util.state

from sbws.util.filelock import FileLock
import os
import json


[docs]class State: ''' State allows one to atomically access and update a simple state file on disk across threads and across processes. To put it blunty, due to limited developer time and his inability to quickly find a way to safely access and update more complex data types (namely, collections like list, set, and dict), you may only store simple types of data as enumerated in _ALLOWED_TYPES. Keys must be strings. Data is stored as JSON on disk in the provided file file. >>> state = State('foo.state') >>> # state == {} >>> state['linux'] = True >>> # 'foo.state' now exists on disk with the JSON for {'linux': True} >>> # We read 'foo.state' from disk in order to get the most up-to-date >>> # state info. Pretend another process has updated 'linux' to be >>> # False >>> state['linux'] >>> # returns False >>> # Pretend another process has added the user's age to the state file. >>> # As before, we read the state file from disk for the most >>> # up-to-date info. >>> state['age'] >>> # Returns 14 >>> # We now set their name. We read the state file first, set the option, >>> # and then write it out. >>> state['name'] = 'John' >>> # We can do many of the same things with a State object as with a dict >>> for key in state: print(key) >>> # Prints 'linux', 'age', and 'name' ''' _ALLOWED_TYPES = (int, float, str, bool, type(None)) def __init__(self, fname): self._fname = fname self._state = self._read() def _read(self): if not os.path.exists(self._fname): return {} with FileLock(self._fname): with open(self._fname, 'rt') as fd: return json.load(fd) def _write(self): with FileLock(self._fname): with open(self._fname, 'wt') as fd: return json.dump(self._state, fd, indent=4) def __len__(self): self._state = self._read() return self._state.__len__()
[docs] def get(self, key, d=None): """ Implements a dictionary ``get`` method reading and locking a json file. """ if not isinstance(key, str): raise TypeError( 'Keys must be strings. %s is a %s' % (key, type(key))) self._state = self._read() return self._state.get(key, d)
def __getitem__(self, key): if not isinstance(key, str): raise TypeError( 'Keys must be strings. %s is a %s' % (key, type(key))) self._state = self._read() return self._state.__getitem__(key) def __delitem__(self, key): if not isinstance(key, str): raise TypeError( 'Keys must be strings. %s is a %s' % (key, type(key))) self._state = self._read() self._state.__delitem__(key) self._write() def __setitem__(self, key, value): if not isinstance(key, str): raise TypeError( 'Keys must be strings. %s is a %s' % (key, type(key))) if type(value) not in State._ALLOWED_TYPES: raise TypeError( 'May only store value with type in %s, not %s' % (State._ALLOWED_TYPES, type(value))) self._state.__setitem__(key, value) self._write() def __iter__(self): self._state = self._read() return self._state.__iter__() def __contains__(self, item): self._state = self._read() return self._state.__contains__(item)