3 from abc import ABC, abstractmethod
9 from typing import Callable, Optional
15 logger = logging.getLogger(__name__)
18 class Persistent(ABC):
20 A base class of an object with a load/save method.
31 def reuse_if_mtime_is_today() -> Callable[[datetime.datetime], bool]:
33 A helper that returns a lambda appropriate for use in the
34 persistent_autoload_singleton decorator's may_reuse_persisted
35 parameter that allows persisted state to be reused as long as it
36 was persisted on the same day as the load.
39 now = datetime.datetime.now()
41 dt.month == now.month and
47 def reuse_if_mtime_less_than_limit(
49 ) -> Callable[[datetime.datetime], bool]:
51 A helper that returns a lambda appropriate for use in the
52 persistent_autoload_singleton decorator's may_reuse_persisted
53 parameter that allows persisted state to be reused as long as it
54 was persisted within the past limit_seconds.
57 now = datetime.datetime.now()
58 return lambda dt: (now - dt).total_seconds() <= limit_seconds
61 class PersistAtShutdown(enum.Enum):
63 An enum to describe the conditions under which state is persisted
64 to disk. See details below.
68 IF_NOT_INITIALIZED_FROM_DISK = 1,
72 class persistent_autoload_singleton(Persistent):
73 """This class is meant to be used as a decorator around a class that:
75 1. Is a singleton; one global instance per python program.
76 2. Has a complex state that is initialized fully by __init__()
77 3. Would benefit from caching said state on disk and reloading
78 it on future invokations rather than recomputing and
81 Here's and example usage pattern:
83 @persistent_autoload_singleton(
84 filename = "my_cache_file.bin",
85 may_reuse_persisted = reuse_if_mtime_less_than_limit(60),
86 persist_at_shutdown = False
88 class MyComplexObject(object):
89 def __init__(self, ...):
90 # do a bunch of work to fully initialize this instance
92 def another_method(self, ...):
93 # use the state stored in this instance to do some work
95 What does this do, exactly?
97 1. Anytime you attempt to instantiate MyComplexObject you will
98 get the same instance. This class is now a singleton.
99 2. The first time you attempt to instantiate MyComplexObject
100 the wrapper scaffolding will check "my_cache_file.bin". If
101 it exists and any may_reuse_persisted predicate indicates
102 that reusing persisted state is allowed, we will skip the
103 call to __init__ and return an unpickled instance read from
105 3. If the file doesn't exist or the predicate indicates that
106 the persisted state cannot be reused, MyComplexObject's
107 __init__ will be invoked and will be expected to fully
108 initialize the instance.
109 4. At program exit time, depending on the value of the
110 persist_at_shutdown parameter, the state of MyComplexObject
111 will be written to disk using the same filename so that
112 future instances may potentially reuse saved state. Note
113 that the state that is persisted is the state at program
121 may_reuse_persisted: Optional[Callable[[datetime.datetime], bool]] = None,
122 persist_at_shutdown: PersistAtShutdown = PersistAtShutdown.NEVER):
123 self.filename = filename
124 self.may_reuse_persisted = may_reuse_persisted
125 self.persist_at_shutdown = persist_at_shutdown
128 def __call__(self, cls):
129 @functools.wraps(cls)
130 def _load(*args, **kwargs):
132 # If class has already been loaded, act like a singleton
133 # and return a reference to the one and only instance in
135 if self.instance is not None:
137 f'Returning already instantiated singleton instance of {cls.__name__}.'
141 was_loaded_from_disk = False
142 if file_utils.does_file_exist(self.filename):
143 cache_mtime_dt = file_utils.get_file_mtime_as_datetime(
146 now = datetime.datetime.now()
148 self.may_reuse_persisted is not None and
149 self.may_reuse_persisted(cache_mtime_dt)
152 f'Attempting to load from persisted cache (mtime={cache_mtime_dt}, {now-cache_mtime_dt} ago)')
154 logger.warning('Loading from cache failed?!')
155 assert self.instance is None
157 assert self.instance is not None
158 was_loaded_from_disk = True
160 if self.instance is None:
162 f'Attempting to instantiate {cls.__name__} directly.'
164 self.instance = cls(*args, **kwargs)
165 was_loaded_from_disk = False
167 assert self.instance is not None
169 self.persist_at_shutdown is PersistAtShutdown.ALWAYS or
171 not was_loaded_from_disk and
172 self.persist_at_shutdown is PersistAtShutdown.IF_NOT_INITIALIZED_FROM_DISK
175 atexit.register(self.save)
179 def load(self) -> bool:
181 with open(self.filename, 'rb') as f:
182 self.instance = dill.load(f)
189 def save(self) -> bool:
190 if self.instance is not None:
192 f'Attempting to save {type(self.instance).__name__} to file {self.filename}'
195 with open(self.filename, 'wb') as f:
196 dill.dump(self.instance, f, dill.HIGHEST_PROTOCOL)