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_sec(
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_sec(60),
86 persist_at_shutdown = PersistAtShutdown.IF_NOT_INITIALIZED_FROM_DISK,
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
104 the disk file. In the example above the predicate allows
105 reuse of saved state if it is <= 60s old.
106 3. If the file doesn't exist or the predicate indicates that
107 the persisted state cannot be reused (e.g. too stale),
108 MyComplexObject's __init__ will be invoked and will be
109 expected to fully initialize the instance.
110 4. At program exit time, depending on the value of the
111 persist_at_shutdown parameter, the state of MyComplexObject
112 will be written to disk using the same filename so that
113 future instances may potentially reuse saved state. Note
114 that the state that is persisted is the state at program
115 exit time. In the example above this parameter indicates
116 that we should persist state so long as we were not
117 initialized from cached state on disk.
124 may_reuse_persisted: Optional[Callable[[datetime.datetime], bool]] = None,
125 persist_at_shutdown: PersistAtShutdown = PersistAtShutdown.NEVER):
126 self.filename = filename
127 self.may_reuse_persisted = may_reuse_persisted
128 self.persist_at_shutdown = persist_at_shutdown
131 def __call__(self, cls):
132 @functools.wraps(cls)
133 def _load(*args, **kwargs):
135 # If class has already been loaded, act like a singleton
136 # and return a reference to the one and only instance in
138 if self.instance is not None:
140 f'Returning already instantiated singleton instance of {cls.__name__}.'
144 was_loaded_from_disk = False
145 if file_utils.does_file_exist(self.filename):
146 cache_mtime_dt = file_utils.get_file_mtime_as_datetime(
149 now = datetime.datetime.now()
151 self.may_reuse_persisted is not None and
152 self.may_reuse_persisted(cache_mtime_dt)
155 f'Attempting to load from persisted cache (mtime={cache_mtime_dt}, {now-cache_mtime_dt} ago)')
157 logger.warning('Loading from cache failed?!')
158 assert self.instance is None
160 assert self.instance is not None
161 was_loaded_from_disk = True
163 if self.instance is None:
165 f'Attempting to instantiate {cls.__name__} directly.'
167 self.instance = cls(*args, **kwargs)
168 was_loaded_from_disk = False
170 assert self.instance is not None
172 self.persist_at_shutdown is PersistAtShutdown.ALWAYS or
174 not was_loaded_from_disk and
175 self.persist_at_shutdown is PersistAtShutdown.IF_NOT_INITIALIZED_FROM_DISK
178 atexit.register(self.save)
182 def load(self) -> bool:
184 with open(self.filename, 'rb') as f:
185 self.instance = dill.load(f)
192 def save(self) -> bool:
193 if self.instance is not None:
195 f'Attempting to save {type(self.instance).__name__} to file {self.filename}'
198 with open(self.filename, 'wb') as f:
199 dill.dump(self.instance, f, dill.HIGHEST_PROTOCOL)