3 # © Copyright 2021-2022, Scott Gasch
5 """A :class:`Persistent` is just a class with a load and save method. This
6 module defines the :class:`Persistent` base and a decorator that can be used to
7 create a persistent singleton that autoloads and autosaves."""
14 from abc import ABC, abstractmethod
15 from typing import Any
19 logger = logging.getLogger(__name__)
22 class Persistent(ABC):
24 A base class of an object with a load/save method. Classes that are
25 decorated with :code:`@persistent_autoloaded_singleton` should subclass
26 this and implement their :meth:`save` and :meth:`load` methods.
30 def save(self) -> bool:
32 Save this thing somewhere that you'll remember when someone calls
33 :meth:`load` later on in a way that makes sense to your code.
40 """Load this thing from somewhere and give back an instance which
41 will become the global singleton and which may (see
42 below) be saved (via :meth:`save`) at program exit time.
44 Oh, in case this is handy, here's a reminder how to write a
45 factory method that doesn't call the c'tor in python::
48 def load_from_somewhere(cls, somewhere):
49 # Note: __new__ does not call __init__.
50 obj = cls.__new__(cls)
52 # Don't forget to call any polymorphic base class initializers
53 super(MyClass, obj).__init__()
55 # Load the piece(s) of obj that you want to from somewhere.
56 obj._state = load_from_somewhere(somewhere)
62 def was_file_written_today(filename: str) -> bool:
63 """Convenience wrapper around :meth:`was_file_written_within_n_seconds`.
66 filename: filename to check
69 True if filename was written today.
72 >>> filename = f'/tmp/testing_persistent_py_{os.getpid()}'
73 >>> os.system(f'touch {filename}')
75 >>> was_file_written_today(filename)
77 >>> os.system(f'touch -d 1974-04-15T01:02:03.99 {filename}')
79 >>> was_file_written_today(filename)
81 >>> os.system(f'/bin/rm -f {filename}')
83 >>> was_file_written_today(filename)
87 if not file_utils.does_file_exist(filename):
90 mtime = file_utils.get_file_mtime_as_datetime(filename)
91 assert mtime is not None
92 now = datetime.datetime.now()
93 return mtime.month == now.month and mtime.day == now.day and mtime.year == now.year
96 def was_file_written_within_n_seconds(
100 """Helper for determining persisted state staleness.
103 filename: the filename to check
104 limit_seconds: how fresh, in seconds, it must be
107 True if filename was written within the past limit_seconds
108 or False otherwise (or on error).
111 >>> filename = f'/tmp/testing_persistent_py_{os.getpid()}'
112 >>> os.system(f'touch {filename}')
114 >>> was_file_written_within_n_seconds(filename, 60)
118 >>> was_file_written_within_n_seconds(filename, 2)
120 >>> os.system(f'/bin/rm -f {filename}')
122 >>> was_file_written_within_n_seconds(filename, 60)
126 if not file_utils.does_file_exist(filename):
129 mtime = file_utils.get_file_mtime_as_datetime(filename)
130 assert mtime is not None
131 now = datetime.datetime.now()
132 return (now - mtime).total_seconds() <= limit_seconds
135 class PersistAtShutdown(enum.Enum):
137 An enum to describe the conditions under which state is persisted
138 to disk. This is passed as an argument to the decorator below and
139 is used to indicate when to call :meth:`save` on a :class:`Persistent`
142 * NEVER: never call :meth:`save`
143 * IF_NOT_LOADED: call :meth:`save` as long as we did not successfully
144 :meth:`load` its state.
145 * ALWAYS: always call :meth:`save`
153 class persistent_autoloaded_singleton(object):
154 """A decorator that can be applied to a :class:`Persistent` subclass
155 (i.e. a class with :meth:`save` and :meth:`load` methods. The
156 decorator will intercept attempts to instantiate the class via
157 it's c'tor and, instead, invoke the class' :meth:`load` to give it a
158 chance to read state from somewhere persistent (disk, db,
159 whatever). Subsequent calls to construt instances of the wrapped
160 class will return a single, global instance (i.e. the wrapped
161 class is a singleton).
163 If :meth:`load` fails (returns None), the c'tor is invoked with the
164 original args as a fallback.
166 Based upon the value of the optional argument
167 :code:`persist_at_shutdown` argument, (NEVER, IF_NOT_LOADED,
168 ALWAYS), the :meth:`save` method of the class will be invoked just
169 before program shutdown to give the class a chance to save its
173 The implementations of :meth:`save` and :meth:`load` and where the
174 class persists its state are details left to the :class:`Persistent`
175 implementation. Essentially this decorator just handles the
176 plumbing of calling your save/load and appropriate times and
177 creates a transparent global singleton whose state can be
178 persisted between runs.
185 persist_at_shutdown: PersistAtShutdown = PersistAtShutdown.IF_NOT_LOADED,
187 self.persist_at_shutdown = persist_at_shutdown
190 def __call__(self, cls: Persistent):
191 @functools.wraps(cls) # type: ignore
192 def _load(*args, **kwargs):
194 # If class has already been loaded, act like a singleton
195 # and return a reference to the one and only instance in
197 if self.instance is not None:
199 'Returning already instantiated singleton instance of %s.', cls.__name__
203 # Otherwise, try to load it from persisted state.
205 logger.debug('Attempting to load %s from persisted state.', cls.__name__)
206 self.instance = cls.load()
207 if not self.instance:
208 msg = 'Loading from cache failed.'
210 logger.debug('Attempting to instantiate %s directly.', cls.__name__)
211 self.instance = cls(*args, **kwargs)
213 logger.debug('Class %s was loaded from persisted state successfully.', cls.__name__)
216 assert self.instance is not None
218 if self.persist_at_shutdown is PersistAtShutdown.ALWAYS or (
219 not was_loaded and self.persist_at_shutdown is PersistAtShutdown.IF_NOT_LOADED
221 logger.debug('Scheduling a deferred called to save at process shutdown time.')
222 atexit.register(self.instance.save)
228 # TODO: PicklingPersistant?
229 # TODO: JsonConfigPersistant?
232 if __name__ == '__main__':