3 # © Copyright 2021-2022, Scott Gasch
5 """A Persistent is just a class with a load and save method. This
6 module defines the 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 @persistent_autoloaded_singleton should subclass this
26 and implement their save() and load() methods.
31 def save(self) -> bool:
33 Save this thing somewhere that you'll remember when someone calls
34 load() later on in a way that makes sense to your code.
42 Load this thing from somewhere and give back an instance which
43 will become the global singleton and which will may (see
44 below) be save()d at program exit time.
46 Oh, in case this is handy, here's how to write a factory
47 method that doesn't call the c'tor in python::
50 def load_from_somewhere(cls, somewhere):
51 # Note: __new__ does not call __init__.
52 obj = cls.__new__(cls)
54 # Don't forget to call any polymorphic base class initializers
55 super(MyClass, obj).__init__()
57 # Load the piece(s) of obj that you want to from somewhere.
58 obj._state = load_from_somewhere(somewhere)
64 def was_file_written_today(filename: str) -> bool:
65 """Returns True if filename was written today.
68 >>> filename = f'/tmp/testing_persistent_py_{os.getpid()}'
69 >>> os.system(f'touch {filename}')
71 >>> was_file_written_today(filename)
73 >>> os.system(f'touch -d 1974-04-15T01:02:03.99 {filename}')
75 >>> was_file_written_today(filename)
77 >>> os.system(f'/bin/rm -f {filename}')
79 >>> was_file_written_today(filename)
83 if not file_utils.does_file_exist(filename):
86 mtime = file_utils.get_file_mtime_as_datetime(filename)
87 assert mtime is not None
88 now = datetime.datetime.now()
89 return mtime.month == now.month and mtime.day == now.day and mtime.year == now.year
92 def was_file_written_within_n_seconds(
96 """Returns True if filename was written within the pas limit_seconds
100 >>> filename = f'/tmp/testing_persistent_py_{os.getpid()}'
101 >>> os.system(f'touch {filename}')
103 >>> was_file_written_within_n_seconds(filename, 60)
107 >>> was_file_written_within_n_seconds(filename, 2)
109 >>> os.system(f'/bin/rm -f {filename}')
111 >>> was_file_written_within_n_seconds(filename, 60)
115 if not file_utils.does_file_exist(filename):
118 mtime = file_utils.get_file_mtime_as_datetime(filename)
119 assert mtime is not None
120 now = datetime.datetime.now()
121 return (now - mtime).total_seconds() <= limit_seconds
124 class PersistAtShutdown(enum.Enum):
126 An enum to describe the conditions under which state is persisted
127 to disk. See details below.
135 class persistent_autoloaded_singleton(object):
136 """A decorator that can be applied to a Persistent subclass (i.e. a
137 class with a save() and load() method. It will intercept attempts
138 to instantiate the class via it's c'tor and, instead, invoke the
139 class' load() method to give it a chance to read state from
140 somewhere persistent.
142 If load() fails (returns None), the c'tor is invoked with the
143 original args as a fallback.
145 Based upon the value of the optional argument persist_at_shutdown,
146 (NEVER, IF_NOT_LOADED, ALWAYS), the save() method of the class will
147 be invoked just before program shutdown to give the class a chance
148 to save its state somewhere.
150 The implementations of save() and load() and where the class
151 persists its state are details left to the Persistent
158 persist_at_shutdown: PersistAtShutdown = PersistAtShutdown.IF_NOT_LOADED,
160 self.persist_at_shutdown = persist_at_shutdown
163 def __call__(self, cls: Persistent):
164 @functools.wraps(cls) # type: ignore
165 def _load(*args, **kwargs):
167 # If class has already been loaded, act like a singleton
168 # and return a reference to the one and only instance in
170 if self.instance is not None:
172 'Returning already instantiated singleton instance of %s.', cls.__name__
176 # Otherwise, try to load it from persisted state.
178 logger.debug('Attempting to load %s from persisted state.', cls.__name__)
179 self.instance = cls.load()
180 if not self.instance:
181 msg = 'Loading from cache failed.'
183 logger.debug('Attempting to instantiate %s directly.', cls.__name__)
184 self.instance = cls(*args, **kwargs)
186 logger.debug('Class %s was loaded from persisted state successfully.', cls.__name__)
189 assert self.instance is not None
191 if self.persist_at_shutdown is PersistAtShutdown.ALWAYS or (
192 not was_loaded and self.persist_at_shutdown is PersistAtShutdown.IF_NOT_LOADED
194 logger.debug('Scheduling a deferred called to save at process shutdown time.')
195 atexit.register(self.instance.save)
201 if __name__ == '__main__':