# © Copyright 2021-2022, Scott Gasch
-"""A :class:`Persistent` is just a class with a load and save method. This
-module defines the :class:`Persistent` base and a decorator that can be used to
-create a persistent singleton that autoloads and autosaves."""
+"""
+This module defines a class hierarchy (base class :class:`Persistent`) and
+a decorator (`@persistent_autoloaded_singleton`) that can be used to create
+objects that load and save their state from some external storage location
+automatically, optionally and conditionally.
+
+A :class:`Persistent` is just a class with a :meth:`Persistent.load` and
+:meth:`Persistent.save` method. Various subclasses such as
+:class:`JsonFileBasedPersistent` and :class:`PicklingFileBasedPersistent`
+define these methods to, save data in a particular format. The details
+of where and whether to save are left to your code to decide by implementing
+interface methods like :meth:`FileBasedPersistent.get_filename` and
+:meth:`FileBasedPersistent.should_we_load_data`.
+
+This module inculdes some helpers to make deciding whether to load persisted
+state easier such as :meth:`was_file_written_today` and
+:meth:`was_file_written_within_n_seconds`.
+
+:class:`Persistent` classes are good for things backed by persisted
+state that is loaded all or most of the time. For example, the high
+score list of a game, the configuration settings of a tool,
+etc... Really anything that wants to save/load state from storage and
+not bother with the plumbing to do so.
+"""
import atexit
import datetime
class FileBasedPersistent(Persistent):
- """A Persistent that uses a file to save/load data and knows the conditions
- under which the state should be saved/loaded."""
+ """A :class:`Persistent` subclass that uses a file to save/load
+ data and knows the conditions under which the state should be
+ saved/loaded.
+ """
@staticmethod
@abstractmethod
def get_filename() -> str:
- """Since this class saves/loads to/from a file, what's its full path?"""
+ """
+ Returns:
+ The full path of the file in which we are saving/loading data.
+ """
pass
@staticmethod
@abstractmethod
def should_we_save_data(filename: str) -> bool:
+ """
+ Returns:
+ True if we should save our state now or False otherwise.
+ """
pass
@staticmethod
@abstractmethod
def should_we_load_data(filename: str) -> bool:
+ """
+ Returns:
+ True if we should load persisted state now or False otherwise.
+ """
pass
@abstractmethod
def get_persistent_data(self) -> Any:
+ """
+ Returns:
+ The raw state data read from the filesystem. Can be any format.
+ """
pass
class PicklingFileBasedPersistent(FileBasedPersistent):
+ """
+ A class that stores its state in a file as pickled Python objects.
+
+ Example usage::
+
+ import persistent
+
+ @persistent.persistent_autoloaded_singleton()
+ class MyClass(persistent.PicklingFileBasedPersistent):
+ def __init__(self, data: Optional[Whatever]):
+ if data:
+ # initialize state from data
+ else:
+ # if desired, initialize an "empty" object with new state.
+
+ @staticmethod
+ @overrides
+ def get_filename() -> str:
+ return "/path/to/where/you/want/to/save/data.bin"
+
+ @staticmethod
+ @overrides
+ def should_we_save_data(filename: str) -> bool:
+ return true_if_we_should_save_the_data_this_time()
+
+ @staticmethod
+ @overrides
+ def should_we_load_data(filename: str) -> bool:
+ return persistent.was_file_written_within_n_seconds(whatever)
+
+ # Persistent will handle the plumbing to instantiate your class from its
+ # persisted state iff the :meth:`should_we_load_data` says it's ok to. It
+ # will also persist the current in-memory state to disk at program exit iff
+ # the :meth:`should_we_save_data` methods says to.
+ c = MyClass()
+
+ """
+
@classmethod
@overrides
def load(cls) -> Optional[Any]:
filename = cls.get_filename()
if cls.should_we_load_data(filename):
- logger.debug('Attempting to load state from %s', filename)
+ logger.debug("Attempting to load state from %s", filename)
assert file_utils.file_is_readable(filename)
import pickle
try:
- with open(filename, 'rb') as rf:
+ with open(filename, "rb") as rf:
data = pickle.load(rf)
return cls(data)
except Exception as e:
- raise Exception(f'Failed to load {filename}.') from e
+ raise Exception(f"Failed to load {filename}.") from e
return None
@overrides
def save(self) -> bool:
filename = self.get_filename()
if self.should_we_save_data(filename):
- logger.debug('Trying to save state in %s', filename)
+ logger.debug("Trying to save state in %s", filename)
try:
import pickle
- with open(filename, 'wb') as wf:
+ with open(filename, "wb") as wf:
pickle.dump(self.get_persistent_data(), wf, pickle.HIGHEST_PROTOCOL)
return True
except Exception as e:
- raise Exception(f'Failed to save to {filename}.') from e
+ raise Exception(f"Failed to save to {filename}.") from e
return False
class JsonFileBasedPersistent(FileBasedPersistent):
+ """A class that stores its state in a JSON format file.
+
+ Example usage::
+
+ import persistent
+
+ @persistent.persistent_autoloaded_singleton()
+ class MyClass(persistent.JsonFileBasedPersistent):
+ def __init__(self, data: Optional[dict[str, Any]]):
+ # load already deserialized the JSON data for you; it's
+ # a "cooked" JSON dict of string -> values, lists, dicts,
+ # etc...
+ if data:
+ #initialize youself from data...
+ else:
+ # if desired, initialize an empty state object
+ # when json_data isn't provided.
+
+ @staticmethod
+ @overrides
+ def get_filename() -> str:
+ return "/path/to/where/you/want/to/save/data.json"
+
+ @staticmethod
+ @overrides
+ def should_we_save_data(filename: str) -> bool:
+ return true_if_we_should_save_the_data_this_time()
+
+ @staticmethod
+ @overrides
+ def should_we_load_data(filename: str) -> bool:
+ return persistent.was_file_written_within_n_seconds(whatever)
+
+ # Persistent will handle the plumbing to instantiate your
+ # class from its persisted state iff the
+ # :meth:`should_we_load_data` says it's ok to. It will also
+ # persist the current in memory state to disk at program exit
+ # iff the :meth:`should_we_save_data methods` says to.
+ c = MyClass()
+ """
+
@classmethod
@overrides
def load(cls) -> Any:
filename = cls.get_filename()
if cls.should_we_load_data(filename):
- logger.debug('Trying to load state from %s', filename)
+ logger.debug("Trying to load state from %s", filename)
import json
try:
- with open(filename, 'r') as rf:
+ with open(filename, "r") as rf:
lines = rf.readlines()
# This is probably bad... but I like comments
# in config files and JSON doesn't support them. So
# pre-process the buffer to remove comments thus
# allowing people to add them.
- buf = ''
+ buf = ""
for line in lines:
- line = re.sub(r'#.*$', '', line)
+ line = re.sub(r"#.*$", "", line)
buf += line
-
json_dict = json.loads(buf)
return cls(json_dict)
except Exception as e:
logger.exception(e)
- raise Exception(f'Failed to load {filename}.') from e
+ raise Exception(f"Failed to load {filename}.") from e
return None
@overrides
def save(self) -> bool:
filename = self.get_filename()
if self.should_we_save_data(filename):
- logger.debug('Trying to save state in %s', filename)
+ logger.debug("Trying to save state in %s", filename)
try:
import json
json_blob = json.dumps(self.get_persistent_data())
- with open(filename, 'w') as wf:
+ with open(filename, "w") as wf:
wf.writelines(json_blob)
return True
except Exception as e:
- raise Exception(f'Failed to save to {filename}.') from e
+ raise Exception(f"Failed to save to {filename}.") from e
return False
"""Convenience wrapper around :meth:`was_file_written_within_n_seconds`.
Args:
- filename: filename to check
+ filename: path / filename to check
Returns:
True if filename was written today.
>>> was_file_written_today(filename)
False
"""
-
if not file_utils.does_file_exist(filename):
return False
chance to read state from somewhere persistent (disk, db,
whatever). Subsequent calls to construt instances of the wrapped
class will return a single, global instance (i.e. the wrapped
- class is a singleton).
+ class is must be a singleton).
If :meth:`load` fails (returns None), the c'tor is invoked with the
original args as a fallback.
# memory.
if self.instance is not None:
logger.debug(
- 'Returning already instantiated singleton instance of %s.',
+ "Returning already instantiated singleton instance of %s.",
cls.__name__,
)
return self.instance
# Otherwise, try to load it from persisted state.
was_loaded = False
- logger.debug('Attempting to load %s from persisted state.', cls.__name__)
+ logger.debug("Attempting to load %s from persisted state.", cls.__name__)
self.instance = cls.load()
if not self.instance:
- msg = 'Loading from cache failed.'
+ msg = "Loading from cache failed."
logger.warning(msg)
- logger.debug('Attempting to instantiate %s directly.', cls.__name__)
+ logger.debug("Attempting to instantiate %s directly.", cls.__name__)
self.instance = cls(*args, **kwargs)
else:
logger.debug(
- 'Class %s was loaded from persisted state successfully.',
+ "Class %s was loaded from persisted state successfully.",
cls.__name__,
)
was_loaded = True
and self.persist_at_shutdown is PersistAtShutdown.IF_NOT_LOADED
):
logger.debug(
- 'Scheduling a deferred called to save at process shutdown time.'
+ "Scheduling a deferred called to save at process shutdown time."
)
atexit.register(self.instance.save)
return self.instance
return _load
-if __name__ == '__main__':
+if __name__ == "__main__":
import doctest
doctest.testmod()
projects / pyutils.git / blobdiff