from abc import ABC, abstractmethod
import atexit
+import datetime
+import enum
import functools
import logging
-
-import dill
+from typing import Any
import file_utils
-
logger = logging.getLogger(__name__)
class Persistent(ABC):
+ """
+ A base class of an object with a load/save method. Classes that are
+ decorated with @persistent_autoloaded_singleton should subclass this
+ and implement their save() and load() methods.
+
+ """
@abstractmethod
- def save(self):
+ def save(self) -> bool:
+ """
+ Save this thing somewhere that you'll remember when someone calls
+ load() later on in a way that makes sense to your code.
+
+ """
pass
+ @classmethod
@abstractmethod
- def load(self):
+ def load(cls) -> Any:
+ """
+ Load this thing from somewhere and give back an instance which
+ will become the global singleton and which will may (see
+ below) be save()d at program exit time.
+
+ Oh, in case this is handy, here's how to write a factory
+ method that doesn't call the c'tor in python:
+
+ @classmethod
+ def load_from_somewhere(cls, somewhere):
+ # Note: __new__ does not call __init__.
+ obj = cls.__new__(cls)
+
+ # Don't forget to call any polymorphic base class initializers
+ super(MyClass, obj).__init__()
+
+ # Load the piece(s) of obj that you want to from somewhere.
+ obj._state = load_from_somewhere(somewhere)
+ return obj
+
+ """
pass
-class persistent_autoload_singleton(Persistent):
- def __init__(self, filename: str, *, max_age_sec: int = 0):
- self.filename = filename
- self.max_age_sec = max_age_sec
+def was_file_written_today(filename: str) -> bool:
+ """Returns True if filename was written today."""
+
+ if not file_utils.does_file_exist(filename):
+ return False
+
+ mtime = file_utils.get_file_mtime_as_datetime(filename)
+ now = datetime.datetime.now()
+ return (
+ mtime.month == now.month and
+ mtime.day == now.day and
+ mtime.year == now.year
+ )
+
+
+def was_file_written_within_n_seconds(
+ filename: str,
+ limit_seconds: int,
+) -> bool:
+ """Returns True if filename was written within the pas limit_seconds
+ seconds.
+
+ """
+ if not file_utils.does_file_exist(filename):
+ return False
+
+ mtime = file_utils.get_file_mtime_as_datetime(filename)
+ now = datetime.datetime.now()
+ return (now - mtime).total_seconds() <= limit_seconds
+
+
+class PersistAtShutdown(enum.Enum):
+ """
+ An enum to describe the conditions under which state is persisted
+ to disk. See details below.
+
+ """
+ NEVER = 0,
+ IF_NOT_LOADED = 1,
+ ALWAYS = 2,
+
+
+class persistent_autoloaded_singleton(object):
+ """A decorator that can be applied to a Persistent subclass (i.e. a
+ class with a save() and load() method. It will intercept attempts
+ to instantiate the class via it's c'tor and, instead, invoke the
+ class' load() method to give it a chance to read state from
+ somewhere persistent.
+
+ If load() fails (returns None), the c'tor is invoked with the
+ original args as a fallback.
+
+ Based upon the value of the optional argument persist_at_shutdown,
+ (NEVER, IF_NOT_LOADED, ALWAYS), the save() method of the class will
+ be invoked just before program shutdown to give the class a chance
+ to save its state somewhere.
+
+ The implementations of save() and load() and where the class
+ persists its state are details left to the Persistent
+ implementation.
+
+ """
+ def __init__(
+ self,
+ *,
+ persist_at_shutdown: PersistAtShutdown = PersistAtShutdown.IF_NOT_LOADED):
+ self.persist_at_shutdown = persist_at_shutdown
self.instance = None
- def __call__(self, cls):
+ def __call__(self, cls: Persistent):
@functools.wraps(cls)
def _load(*args, **kwargs):
)
return self.instance
- if not self.load():
- assert self.instance is None
- logger.debug(
- f'Instantiating {cls.__name__} directly.'
- )
+ # Otherwise, try to load it from persisted state.
+ was_loaded = False
+ logger.debug(f'Attempting to load {cls.__name__} from persisted state.')
+ self.instance = cls.load()
+ if not self.instance:
+ logger.warning('Loading from cache failed.')
+ logger.debug(f'Attempting to instantiate {cls.__name__} directly.')
self.instance = cls(*args, **kwargs)
+ else:
+ logger.debug(f'Class {cls.__name__} was loaded from persisted state successfully.')
+ was_loaded = True
- # On program exit, save state to disk.
- atexit.register(self.save)
assert self.instance is not None
+
+ if (
+ self.persist_at_shutdown is PersistAtShutdown.ALWAYS or
+ (
+ not was_loaded and
+ self.persist_at_shutdown is PersistAtShutdown.IF_NOT_LOADED
+ )
+ ):
+ logger.debug('Scheduling a deferred called to save at process shutdown time.')
+ atexit.register(self.instance.save)
return self.instance
return _load
-
- def load(self) -> bool:
- if (
- file_utils.does_file_exist(self.filename)
- and (
- self.max_age_sec == 0 or
- file_utils.get_file_mtime_age_seconds(self.filename) <= self.max_age_sec
- )
- ):
- logger.debug(
- f'Attempting to load from file {self.filename}'
- )
- try:
- with open(self.filename, 'rb') as f:
- self.instance = dill.load(f)
- return True
- except Exception:
- self.instance = None
- return False
- return False
-
- def save(self) -> bool:
- if self.instance is not None:
- logger.debug(
- f'Attempting to save {type(self.instance).__name__} to file {self.filename}'
- )
- try:
- with open(self.filename, 'wb') as f:
- dill.dump(self.instance, f, dill.HIGHEST_PROTOCOL)
- return True
- except Exception:
- return False
- return False
projects / python_utils.git / blobdiff