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.
43 Load this thing from somewhere and give back an instance which
44 will become the global singleton and which will may (see
45 below) be save()d at program exit time.
47 Oh, in case this is handy, here's how to write a factory
48 method that doesn't call the c'tor in python:
51 def load_from_somewhere(cls, somewhere):
52 # Note: __new__ does not call __init__.
53 obj = cls.__new__(cls)
55 # Don't forget to call any polymorphic base class initializers
56 super(MyClass, obj).__init__()
58 # Load the piece(s) of obj that you want to from somewhere.
59 obj._state = load_from_somewhere(somewhere)
66 def was_file_written_today(filename: str) -> bool:
67 """Returns True if filename was written today.
70 >>> filename = f'/tmp/testing_persistent_py_{os.getpid()}'
71 >>> os.system(f'touch {filename}')
73 >>> was_file_written_today(filename)
75 >>> os.system(f'touch -d 1974-04-15T01:02:03.99 {filename}')
77 >>> was_file_written_today(filename)
79 >>> os.system(f'/bin/rm -f {filename}')
81 >>> was_file_written_today(filename)
85 if not file_utils.does_file_exist(filename):
88 mtime = file_utils.get_file_mtime_as_datetime(filename)
89 assert mtime is not None
90 now = datetime.datetime.now()
91 return mtime.month == now.month and mtime.day == now.day and mtime.year == now.year
94 def was_file_written_within_n_seconds(
98 """Returns True if filename was written within the pas limit_seconds
102 >>> filename = f'/tmp/testing_persistent_py_{os.getpid()}'
103 >>> os.system(f'touch {filename}')
105 >>> was_file_written_within_n_seconds(filename, 60)
109 >>> was_file_written_within_n_seconds(filename, 2)
111 >>> os.system(f'/bin/rm -f {filename}')
113 >>> was_file_written_within_n_seconds(filename, 60)
117 if not file_utils.does_file_exist(filename):
120 mtime = file_utils.get_file_mtime_as_datetime(filename)
121 assert mtime is not None
122 now = datetime.datetime.now()
123 return (now - mtime).total_seconds() <= limit_seconds
126 class PersistAtShutdown(enum.Enum):
128 An enum to describe the conditions under which state is persisted
129 to disk. See details below.
138 class persistent_autoloaded_singleton(object):
139 """A decorator that can be applied to a Persistent subclass (i.e. a
140 class with a save() and load() method. It will intercept attempts
141 to instantiate the class via it's c'tor and, instead, invoke the
142 class' load() method to give it a chance to read state from
143 somewhere persistent.
145 If load() fails (returns None), the c'tor is invoked with the
146 original args as a fallback.
148 Based upon the value of the optional argument persist_at_shutdown,
149 (NEVER, IF_NOT_LOADED, ALWAYS), the save() method of the class will
150 be invoked just before program shutdown to give the class a chance
151 to save its state somewhere.
153 The implementations of save() and load() and where the class
154 persists its state are details left to the Persistent
162 persist_at_shutdown: PersistAtShutdown = PersistAtShutdown.IF_NOT_LOADED,
164 self.persist_at_shutdown = persist_at_shutdown
167 def __call__(self, cls: Persistent):
168 @functools.wraps(cls) # type: ignore
169 def _load(*args, **kwargs):
171 # If class has already been loaded, act like a singleton
172 # and return a reference to the one and only instance in
174 if self.instance is not None:
176 'Returning already instantiated singleton instance of %s.', cls.__name__
180 # Otherwise, try to load it from persisted state.
182 logger.debug('Attempting to load %s from persisted state.', cls.__name__)
183 self.instance = cls.load()
184 if not self.instance:
185 msg = 'Loading from cache failed.'
187 logger.debug('Attempting to instantiate %s directly.', cls.__name__)
188 self.instance = cls(*args, **kwargs)
190 logger.debug('Class %s was loaded from persisted state successfully.', cls.__name__)
193 assert self.instance is not None
195 if self.persist_at_shutdown is PersistAtShutdown.ALWAYS or (
196 not was_loaded and self.persist_at_shutdown is PersistAtShutdown.IF_NOT_LOADED
198 logger.debug('Scheduling a deferred called to save at process shutdown time.')
199 atexit.register(self.instance.save)
205 if __name__ == '__main__':