3 from abc import ABC, abstractmethod
14 logger = logging.getLogger(__name__)
17 class Persistent(ABC):
19 A base class of an object with a load/save method. Classes that are
20 decorated with @persistent_autoloaded_singleton should subclass this
21 and implement their save() and load() methods.
25 def save(self) -> bool:
27 Save this thing somewhere that you'll remember when someone calls
28 load() later on in a way that makes sense to your code.
37 Load this thing from somewhere and give back an instance which
38 will become the global singleton and which will may (see
39 below) be save()d at program exit time.
41 Oh, in case this is handy, here's how to write a factory
42 method that doesn't call the c'tor in python:
45 def load_from_somewhere(cls, somewhere):
46 # Note: __new__ does not call __init__.
47 obj = cls.__new__(cls)
49 # Don't forget to call any polymorphic base class initializers
50 super(MyClass, obj).__init__()
52 # Load the piece(s) of obj that you want to from somewhere.
53 obj._state = load_from_somewhere(somewhere)
60 def was_file_written_today(filename: str) -> bool:
61 """Returns True if filename was written today."""
63 if not file_utils.does_file_exist(filename):
66 mtime = file_utils.get_file_mtime_as_datetime(filename)
67 now = datetime.datetime.now()
69 mtime.month == now.month and
70 mtime.day == now.day and
71 mtime.year == now.year
75 def was_file_written_within_n_seconds(
79 """Returns True if filename was written within the pas limit_seconds
83 if not file_utils.does_file_exist(filename):
86 mtime = file_utils.get_file_mtime_as_datetime(filename)
87 now = datetime.datetime.now()
88 return (now - mtime).total_seconds() <= limit_seconds
91 class PersistAtShutdown(enum.Enum):
93 An enum to describe the conditions under which state is persisted
94 to disk. See details below.
102 class persistent_autoloaded_singleton(object):
103 """A decorator that can be applied to a Persistent subclass (i.e. a
104 class with a save() and load() method. It will intercept attempts
105 to instantiate the class via it's c'tor and, instead, invoke the
106 class' load() method to give it a chance to read state from
107 somewhere persistent.
109 If load() fails (returns None), the c'tor is invoked with the
110 original args as a fallback.
112 Based upon the value of the optional argument persist_at_shutdown,
113 (NEVER, IF_NOT_LOADED, ALWAYS), the save() method of the class will
114 be invoked just before program shutdown to give the class a chance
115 to save its state somewhere.
117 The implementations of save() and load() and where the class
118 persists its state are details left to the Persistent
125 persist_at_shutdown: PersistAtShutdown = PersistAtShutdown.IF_NOT_LOADED):
126 self.persist_at_shutdown = persist_at_shutdown
129 def __call__(self, cls: Persistent):
130 @functools.wraps(cls)
131 def _load(*args, **kwargs):
133 # If class has already been loaded, act like a singleton
134 # and return a reference to the one and only instance in
136 if self.instance is not None:
138 f'Returning already instantiated singleton instance of {cls.__name__}.'
142 # Otherwise, try to load it from persisted state.
144 logger.debug(f'Attempting to load {cls.__name__} from persisted state.')
145 self.instance = cls.load()
146 if not self.instance:
147 msg = 'Loading from cache failed.'
149 logger.debug(f'Attempting to instantiate {cls.__name__} directly.')
150 self.instance = cls(*args, **kwargs)
152 logger.debug(f'Class {cls.__name__} was loaded from persisted state successfully.')
155 assert self.instance is not None
158 self.persist_at_shutdown is PersistAtShutdown.ALWAYS or
161 self.persist_at_shutdown is PersistAtShutdown.IF_NOT_LOADED
164 logger.debug('Scheduling a deferred called to save at process shutdown time.')
165 atexit.register(self.instance.save)