Source code for sbws.util.state
import json
import os
from sbws.util.filelock import FileLock
from .json import CustomDecoder, CustomEncoder
[docs]
class State:
"""
`json` wrapper to read a json file every time it gets a key and to write
to the file every time a key is set.
Every time a key is got or set, the file is locked, to atomically access
and update the file across threads and across processes.
>>> 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'
"""
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, cls=CustomDecoder)
def _write(self):
with FileLock(self._fname):
with open(self._fname, "wt") as fd:
return json.dump(self._state, fd, indent=4, cls=CustomEncoder)
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.
"""
self._state = self._read()
return self._state.get(key, d)
def __getitem__(self, key):
self._state = self._read()
return self._state.__getitem__(key)
def __delitem__(self, key):
self._state = self._read()
self._state.__delitem__(key)
self._write()
def __setitem__(self, key, value):
# NOTE: important, read the file before setting the key,
# otherwise if other instances are creating other keys, they're lost.
self._state = self._read()
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)
[docs]
def count(self, k):
"""
Returns the length if the key value is a list
or the sum of number if the key value is a list of list
or the key value
or None if the state doesn't have the key.
"""
if self.get(k):
if isinstance(self._state[k], list):
if isinstance(self._state[k][0], list):
return sum(map(lambda x: x[1], self._state[k]))
return len(self._state[k])
return self.get(k)
return None