3 # © Copyright 2021-2023, Scott Gasch
6 This module defines a class hierarchy (base class :class:`Persistent`) and
7 a decorator (`@persistent_autoloaded_singleton`) that can be used to create
8 objects that load and save their state from some external storage location
9 automatically, optionally and conditionally.
11 A :class:`Persistent` is just a class with a :meth:`Persistent.load` and
12 :meth:`Persistent.save` method. Various subclasses such as
13 :class:`JsonFileBasedPersistent` and :class:`PicklingFileBasedPersistent`
14 define these methods to, save data in a particular format. The details
15 of where and whether to save are left to your code to decide by implementing
16 interface methods like :meth:`FileBasedPersistent.get_filename` and
17 :meth:`FileBasedPersistent.should_we_load_data`.
19 This module inculdes some helpers to make deciding whether to load persisted
20 state easier such as :meth:`was_file_written_today` and
21 :meth:`was_file_written_within_n_seconds`.
23 :class:`Persistent` classes are good for things backed by persisted
24 state that is loaded all or most of the time. For example, the high
25 score list of a game, the configuration settings of a tool,
26 etc... Really anything that wants to save/load state from storage and
27 not bother with the plumbing to do so.
30 from __future__ import annotations
38 from abc import ABC, abstractmethod
39 from typing import Any, Optional
41 from overrides import overrides
43 from pyutils.files import file_utils
45 logger = logging.getLogger(__name__)
48 class Persistent(ABC):
50 A base class of an object with a load/save method. Classes that are
51 decorated with :code:`@persistent_autoloaded_singleton` should subclass
52 this and implement their :meth:`save` and :meth:`load` methods.
56 def save(self) -> bool:
58 Save this thing somewhere that you'll remember when someone calls
59 :meth:`load` later on in a way that makes sense to your code.
65 def load(cls: type[Persistent]) -> Optional[Persistent]:
66 """Load this thing from somewhere and give back an instance which
67 will become the global singleton and which may (see
68 below) be saved (via :meth:`save`) at program exit time.
70 Oh, in case this is handy, here's a reminder how to write a
71 factory method that doesn't call the c'tor in python::
74 def load_from_somewhere(cls, somewhere):
75 # Note: __new__ does not call __init__.
76 obj = cls.__new__(cls)
78 # Don't forget to call any polymorphic base class initializers
79 super(MyClass, obj).__init__()
81 # Load the piece(s) of obj that you want to from somewhere.
82 obj._state = load_from_somewhere(somewhere)
86 cls: the class (type) that is being instantiated. That is, the
90 An instance of the requested type or None to indicate failure.
96 class FileBasedPersistent(Persistent):
97 """A :class:`Persistent` subclass that uses a file to save/load
98 data and knows the conditions under which the state should be
104 def get_filename() -> str:
107 The full path of the file in which we are saving/loading data.
113 def should_we_save_data(filename: str) -> bool:
116 True if we should save our state now or False otherwise.
122 def should_we_load_data(filename: str) -> bool:
125 True if we should load persisted state now or False otherwise.
130 def get_persistent_data(self) -> Any:
133 The raw state data read from the filesystem. Can be any format.
138 class PicklingFileBasedPersistent(FileBasedPersistent):
140 A class that stores its state in a file as pickled Python objects.
144 from pyutils.typez import persistent
146 @persistent.persistent_autoloaded_singleton()
147 class MyClass(persistent.PicklingFileBasedPersistent):
148 def __init__(self, data: Optional[Whatever]):
150 # initialize state from data
152 # if desired, initialize an "empty" object with new state.
156 def get_filename() -> str:
157 return "/path/to/where/you/want/to/save/data.bin"
161 def should_we_save_data(filename: str) -> bool:
162 return true_if_we_should_save_the_data_this_time()
166 def should_we_load_data(filename: str) -> bool:
167 return persistent.was_file_written_within_n_seconds(whatever)
169 # Persistent will handle the plumbing to instantiate your class from its
170 # persisted state iff the :meth:`should_we_load_data` says it's ok to. It
171 # will also persist the current in-memory state to disk at program exit iff
172 # the :meth:`should_we_save_data` methods says to.
178 def __init__(self, data: Optional[Any] = None):
179 """You should override this."""
185 cls: type[PicklingFileBasedPersistent],
186 ) -> Optional[PicklingFileBasedPersistent]:
189 Exception: failure to load from file.
191 filename = cls.get_filename()
192 if cls.should_we_load_data(filename):
193 logger.debug("Attempting to load state from %s", filename)
194 assert file_utils.is_readable(filename)
199 with open(filename, mode="rb") as rf:
200 data = pickle.load(rf)
203 except Exception as e:
204 raise Exception(f"Failed to load {filename}.") from e
208 def save(self) -> bool:
211 Exception: failure to save to file.
213 filename = self.get_filename()
214 if self.should_we_save_data(filename):
215 logger.debug("Trying to save state in %s", filename)
219 with file_utils.CreateFileWithMode(filename, 0o600, "wb") as wf:
220 pickle.dump(self.get_persistent_data(), wf, pickle.HIGHEST_PROTOCOL)
222 except Exception as e:
223 raise Exception(f"Failed to save to {filename}.") from e
227 class JsonFileBasedPersistent(FileBasedPersistent):
228 """A class that stores its state in a JSON format file.
232 from pyutils.typez import persistent
234 @persistent.persistent_autoloaded_singleton()
235 class MyClass(persistent.JsonFileBasedPersistent):
236 def __init__(self, data: Optional[dict[str, Any]]):
237 # load already deserialized the JSON data for you; it's
238 # a "cooked" JSON dict of string -> values, lists, dicts,
241 #initialize youself from data...
243 # if desired, initialize an empty state object
244 # when json_data isn't provided.
248 def get_filename() -> str:
249 return "/path/to/where/you/want/to/save/data.json"
253 def should_we_save_data(filename: str) -> bool:
254 return true_if_we_should_save_the_data_this_time()
258 def should_we_load_data(filename: str) -> bool:
259 return persistent.was_file_written_within_n_seconds(whatever)
261 # Persistent will handle the plumbing to instantiate your
262 # class from its persisted state iff the
263 # :meth:`should_we_load_data` says it's ok to. It will also
264 # persist the current in memory state to disk at program exit
265 # iff the :meth:`should_we_save_data methods` says to.
270 def __init__(self, data: Optional[Any]):
271 """You should override this."""
276 def load(cls: type[JsonFileBasedPersistent]) -> Optional[JsonFileBasedPersistent]:
279 Exception: failure to load from file.
281 filename = cls.get_filename()
282 if cls.should_we_load_data(filename):
283 logger.debug("Trying to load state from %s", filename)
287 with open(filename, mode="r", encoding="utf-8") as rf:
288 lines = rf.readlines()
290 # This is probably bad... but I like comments
291 # in config files and JSON doesn't support them. So
292 # pre-process the buffer to remove comments thus
293 # allowing people to add them.
296 line = re.sub(r"#.*$", "", line)
298 json_dict = json.loads(buf)
299 return cls(json_dict)
301 except Exception as e:
303 "Failed to load path %s; raising an exception", filename
305 raise Exception(f"Failed to load {filename}.") from e
309 def save(self) -> bool:
312 Exception: failure to save to file.
314 filename = self.get_filename()
315 if self.should_we_save_data(filename):
316 logger.debug("Trying to save state in %s", filename)
320 json_blob = json.dumps(self.get_persistent_data())
321 with open(filename, mode="w", encoding="utf-8") as wf:
322 wf.writelines(json_blob)
324 except Exception as e:
325 raise Exception(f"Failed to save to {filename}.") from e
329 def was_file_written_today(filename: str) -> bool:
330 """Convenience wrapper around :meth:`was_file_written_within_n_seconds`.
333 filename: path / filename to check
336 True if filename was written today.
339 >>> filename = f'/tmp/testing_persistent_py_{os.getpid()}'
340 >>> os.system(f'touch {filename}')
342 >>> was_file_written_today(filename)
344 >>> os.system(f'touch -d 1974-04-15T01:02:03.99 {filename}')
346 >>> was_file_written_today(filename)
348 >>> os.system(f'/bin/rm -f {filename}')
350 >>> was_file_written_today(filename)
353 if not file_utils.does_file_exist(filename):
356 mtime = file_utils.get_file_mtime_as_datetime(filename)
357 assert mtime is not None
358 now = datetime.datetime.now()
359 return mtime.month == now.month and mtime.day == now.day and mtime.year == now.year
362 def was_file_written_within_n_seconds(
366 """Helper for determining persisted state staleness.
369 filename: the filename to check
370 limit_seconds: how fresh, in seconds, it must be
373 True if filename was written within the past limit_seconds
374 or False otherwise (or on error).
377 >>> filename = f'/tmp/testing_persistent_py_{os.getpid()}'
378 >>> os.system(f'touch {filename}')
380 >>> was_file_written_within_n_seconds(filename, 60)
384 >>> was_file_written_within_n_seconds(filename, 2)
386 >>> os.system(f'/bin/rm -f {filename}')
388 >>> was_file_written_within_n_seconds(filename, 60)
392 if not file_utils.does_file_exist(filename):
395 mtime = file_utils.get_file_mtime_as_datetime(filename)
396 assert mtime is not None
397 now = datetime.datetime.now()
398 return (now - mtime).total_seconds() <= limit_seconds
401 class PersistAtShutdown(enum.Enum):
403 An enum to describe the conditions under which state is persisted
404 to disk. This is passed as an argument to the decorator below and
405 is used to indicate when to call :meth:`save` on a :class:`Persistent`
408 * NEVER: never call :meth:`save`
409 * IF_NOT_LOADED: call :meth:`save` as long as we did not successfully
410 :meth:`load` its state.
411 * ALWAYS: always call :meth:`save`
419 class persistent_autoloaded_singleton(object):
420 """A decorator that can be applied to a :class:`Persistent` subclass
421 (i.e. a class with :meth:`save` and :meth:`load` methods. The
422 decorator will intercept attempts to instantiate the class via
423 it's c'tor and, instead, invoke the class' :meth:`load` to give it a
424 chance to read state from somewhere persistent (disk, db,
425 whatever). Subsequent calls to construct instances of the wrapped
426 class will return a single, global instance (i.e. the wrapped
427 class is must be a singleton).
429 If :meth:`load` fails (returns None), the class' c'tor is invoked
430 with the original args as a fallback.
432 Based upon the value of the optional argument
433 :code:`persist_at_shutdown` argument, (NEVER, IF_NOT_LOADED,
434 ALWAYS), the :meth:`save` method of the class will be invoked just
435 before program shutdown to give the class a chance to save its
439 The implementations of :meth:`save` and :meth:`load` and where the
440 class persists its state are details left to the :class:`Persistent`
441 implementation. Essentially this decorator just handles the
442 plumbing of calling your save/load and appropriate times and
443 creates a transparent global singleton whose state can be
444 persisted between runs. See example implementations such as
445 :class:`JsonFileBasedPersistent` and
446 :class:`PicklingFileBasedPersistent`.
452 persist_at_shutdown: PersistAtShutdown = PersistAtShutdown.IF_NOT_LOADED,
454 self.persist_at_shutdown = persist_at_shutdown
457 def __call__(self, cls: Persistent):
458 @functools.wraps(cls) # type: ignore
459 def _load(*args, **kwargs):
461 # If class has already been loaded, act like a singleton
462 # and return a reference to the one and only instance in
464 if self.instance is not None:
466 "Returning already instantiated singleton instance of %s.",
471 # Otherwise, try to load it from persisted state.
473 logger.debug("Attempting to load %s from persisted state.", cls.__name__)
474 self.instance = cls.load()
475 if not self.instance:
476 msg = "Loading from cache failed."
478 logger.debug("Attempting to instantiate %s directly.", cls.__name__)
479 self.instance = cls(*args, **kwargs)
482 "Class %s was loaded from persisted state successfully.",
487 assert self.instance is not None
489 if self.persist_at_shutdown is PersistAtShutdown.ALWAYS or (
491 and self.persist_at_shutdown is PersistAtShutdown.IF_NOT_LOADED
494 "Scheduling a deferred called to save at process shutdown time."
496 atexit.register(self.instance.save)
502 if __name__ == "__main__":