persistent.py

   1 #!/usr/bin/env python3
   2
   3 """A Persistent is just a class with a load and save method.  This
   4 module defines the Persistent base and a decorator that can be used to
   5 create a persistent singleton that autoloads and autosaves."""
   6
   7 import atexit
   8 import datetime
   9 import enum
  10 import functools
  11 import logging
  12 from abc import ABC, abstractmethod
  13 from typing import Any
  14
  15 import file_utils
  16
  17 logger = logging.getLogger(__name__)
  18
  19
  20 class Persistent(ABC):
  21     """
  22     A base class of an object with a load/save method.  Classes that are
  23     decorated with @persistent_autoloaded_singleton should subclass this
  24     and implement their save() and load() methods.
  25
  26     """
  27
  28     @abstractmethod
  29     def save(self) -> bool:
  30         """
  31         Save this thing somewhere that you'll remember when someone calls
  32         load() later on in a way that makes sense to your code.
  33
  34         """
  35         pass
  36
  37     @classmethod
  38     @abstractmethod
  39     def load(cls) -> Any:
  40         """
  41         Load this thing from somewhere and give back an instance which
  42         will become the global singleton and which will may (see
  43         below) be save()d at program exit time.
  44
  45         Oh, in case this is handy, here's how to write a factory
  46         method that doesn't call the c'tor in python:
  47
  48             @classmethod
  49             def load_from_somewhere(cls, somewhere):
  50                 # Note: __new__ does not call __init__.
  51                 obj = cls.__new__(cls)
  52
  53                 # Don't forget to call any polymorphic base class initializers
  54                 super(MyClass, obj).__init__()
  55
  56                 # Load the piece(s) of obj that you want to from somewhere.
  57                 obj._state = load_from_somewhere(somewhere)
  58                 return obj
  59
  60         """
  61         pass
  62
  63
  64 def was_file_written_today(filename: str) -> bool:
  65     """Returns True if filename was written today."""
  66
  67     if not file_utils.does_file_exist(filename):
  68         return False
  69
  70     mtime = file_utils.get_file_mtime_as_datetime(filename)
  71     assert mtime is not None
  72     now = datetime.datetime.now()
  73     return mtime.month == now.month and mtime.day == now.day and mtime.year == now.year
  74
  75
  76 def was_file_written_within_n_seconds(
  77     filename: str,
  78     limit_seconds: int,
  79 ) -> bool:
  80     """Returns True if filename was written within the pas limit_seconds
  81     seconds.
  82
  83     """
  84     if not file_utils.does_file_exist(filename):
  85         return False
  86
  87     mtime = file_utils.get_file_mtime_as_datetime(filename)
  88     assert mtime is not None
  89     now = datetime.datetime.now()
  90     return (now - mtime).total_seconds() <= limit_seconds
  91
  92
  93 class PersistAtShutdown(enum.Enum):
  94     """
  95     An enum to describe the conditions under which state is persisted
  96     to disk.  See details below.
  97
  98     """
  99
 100     NEVER = (0,)
 101     IF_NOT_LOADED = (1,)
 102     ALWAYS = (2,)
 103
 104
 105 class persistent_autoloaded_singleton(object):
 106     """A decorator that can be applied to a Persistent subclass (i.e.  a
 107     class with a save() and load() method.  It will intercept attempts
 108     to instantiate the class via it's c'tor and, instead, invoke the
 109     class' load() method to give it a chance to read state from
 110     somewhere persistent.
 111
 112     If load() fails (returns None), the c'tor is invoked with the
 113     original args as a fallback.
 114
 115     Based upon the value of the optional argument persist_at_shutdown,
 116     (NEVER, IF_NOT_LOADED, ALWAYS), the save() method of the class will
 117     be invoked just before program shutdown to give the class a chance
 118     to save its state somewhere.
 119
 120     The implementations of save() and load() and where the class
 121     persists its state are details left to the Persistent
 122     implementation.
 123
 124     """
 125
 126     def __init__(
 127         self,
 128         *,
 129         persist_at_shutdown: PersistAtShutdown = PersistAtShutdown.IF_NOT_LOADED,
 130     ):
 131         self.persist_at_shutdown = persist_at_shutdown
 132         self.instance = None
 133
 134     def __call__(self, cls: Persistent):
 135         @functools.wraps(cls)  # type: ignore
 136         def _load(*args, **kwargs):
 137
 138             # If class has already been loaded, act like a singleton
 139             # and return a reference to the one and only instance in
 140             # memory.
 141             if self.instance is not None:
 142                 logger.debug(
 143                     'Returning already instantiated singleton instance of %s.', cls.__name__
 144                 )
 145                 return self.instance
 146
 147             # Otherwise, try to load it from persisted state.
 148             was_loaded = False
 149             logger.debug('Attempting to load %s from persisted state.', cls.__name__)
 150             self.instance = cls.load()
 151             if not self.instance:
 152                 msg = 'Loading from cache failed.'
 153                 logger.warning(msg)
 154                 logger.debug('Attempting to instantiate %s directly.', cls.__name__)
 155                 self.instance = cls(*args, **kwargs)
 156             else:
 157                 logger.debug('Class %s was loaded from persisted state successfully.', cls.__name__)
 158                 was_loaded = True
 159
 160             assert self.instance is not None
 161
 162             if self.persist_at_shutdown is PersistAtShutdown.ALWAYS or (
 163                 not was_loaded and self.persist_at_shutdown is PersistAtShutdown.IF_NOT_LOADED
 164             ):
 165                 logger.debug('Scheduling a deferred called to save at process shutdown time.')
 166                 atexit.register(self.instance.save)
 167             return self.instance
 168
 169         return _load