persistent.py

   1 #!/usr/bin/env python3
   2
   3 # © Copyright 2021-2022, Scott Gasch
   4
   5 """A :class:`Persistent` is just a class with a load and save method.  This
   6 module defines the :class:`Persistent` base and a decorator that can be used to
   7 create a persistent singleton that autoloads and autosaves."""
   8
   9 import atexit
  10 import datetime
  11 import enum
  12 import functools
  13 import logging
  14 from abc import ABC, abstractmethod
  15 from typing import Any
  16
  17 import file_utils
  18
  19 logger = logging.getLogger(__name__)
  20
  21
  22 class Persistent(ABC):
  23     """
  24     A base class of an object with a load/save method.  Classes that are
  25     decorated with :code:`@persistent_autoloaded_singleton` should subclass
  26     this and implement their :meth:`save` and :meth:`load` methods.
  27     """
  28
  29     @abstractmethod
  30     def save(self) -> bool:
  31         """
  32         Save this thing somewhere that you'll remember when someone calls
  33         :meth:`load` later on in a way that makes sense to your code.
  34         """
  35         pass
  36
  37     @classmethod
  38     @abstractmethod
  39     def load(cls) -> Any:
  40         """Load this thing from somewhere and give back an instance which
  41         will become the global singleton and which may (see
  42         below) be saved (via :meth:`save`) at program exit time.
  43
  44         Oh, in case this is handy, here's a reminder how to write a
  45         factory method that doesn't call the c'tor in python::
  46
  47             @classmethod
  48             def load_from_somewhere(cls, somewhere):
  49                 # Note: __new__ does not call __init__.
  50                 obj = cls.__new__(cls)
  51
  52                 # Don't forget to call any polymorphic base class initializers
  53                 super(MyClass, obj).__init__()
  54
  55                 # Load the piece(s) of obj that you want to from somewhere.
  56                 obj._state = load_from_somewhere(somewhere)
  57                 return obj
  58         """
  59         pass
  60
  61
  62 def was_file_written_today(filename: str) -> bool:
  63     """Convenience wrapper around :meth:`was_file_written_within_n_seconds`.
  64
  65     Args:
  66         filename: filename to check
  67
  68     Returns:
  69         True if filename was written today.
  70
  71     >>> import os
  72     >>> filename = f'/tmp/testing_persistent_py_{os.getpid()}'
  73     >>> os.system(f'touch {filename}')
  74     0
  75     >>> was_file_written_today(filename)
  76     True
  77     >>> os.system(f'touch -d 1974-04-15T01:02:03.99 {filename}')
  78     0
  79     >>> was_file_written_today(filename)
  80     False
  81     >>> os.system(f'/bin/rm -f {filename}')
  82     0
  83     >>> was_file_written_today(filename)
  84     False
  85     """
  86
  87     if not file_utils.does_file_exist(filename):
  88         return False
  89
  90     mtime = file_utils.get_file_mtime_as_datetime(filename)
  91     assert mtime is not None
  92     now = datetime.datetime.now()
  93     return mtime.month == now.month and mtime.day == now.day and mtime.year == now.year
  94
  95
  96 def was_file_written_within_n_seconds(
  97     filename: str,
  98     limit_seconds: int,
  99 ) -> bool:
 100     """Helper for determining persisted state staleness.
 101
 102     Args:
 103         filename: the filename to check
 104         limit_seconds: how fresh, in seconds, it must be
 105
 106     Returns:
 107         True if filename was written within the past limit_seconds
 108         or False otherwise (or on error).
 109
 110     >>> import os
 111     >>> filename = f'/tmp/testing_persistent_py_{os.getpid()}'
 112     >>> os.system(f'touch {filename}')
 113     0
 114     >>> was_file_written_within_n_seconds(filename, 60)
 115     True
 116     >>> import time
 117     >>> time.sleep(2.0)
 118     >>> was_file_written_within_n_seconds(filename, 2)
 119     False
 120     >>> os.system(f'/bin/rm -f {filename}')
 121     0
 122     >>> was_file_written_within_n_seconds(filename, 60)
 123     False
 124     """
 125
 126     if not file_utils.does_file_exist(filename):
 127         return False
 128
 129     mtime = file_utils.get_file_mtime_as_datetime(filename)
 130     assert mtime is not None
 131     now = datetime.datetime.now()
 132     return (now - mtime).total_seconds() <= limit_seconds
 133
 134
 135 class PersistAtShutdown(enum.Enum):
 136     """
 137     An enum to describe the conditions under which state is persisted
 138     to disk.  This is passed as an argument to the decorator below and
 139     is used to indicate when to call :meth:`save` on a :class:`Persistent`
 140     subclass.
 141
 142     * NEVER: never call :meth:`save`
 143     * IF_NOT_LOADED: call :meth:`save` as long as we did not successfully
 144       :meth:`load` its state.
 145     * ALWAYS: always call :meth:`save`
 146     """
 147
 148     NEVER = (0,)
 149     IF_NOT_LOADED = (1,)
 150     ALWAYS = (2,)
 151
 152
 153 class persistent_autoloaded_singleton(object):
 154     """A decorator that can be applied to a :class:`Persistent` subclass
 155     (i.e.  a class with :meth:`save` and :meth:`load` methods.  The
 156     decorator will intercept attempts to instantiate the class via
 157     it's c'tor and, instead, invoke the class' :meth:`load` to give it a
 158     chance to read state from somewhere persistent (disk, db,
 159     whatever).  Subsequent calls to construt instances of the wrapped
 160     class will return a single, global instance (i.e. the wrapped
 161     class is a singleton).
 162
 163     If :meth:`load` fails (returns None), the c'tor is invoked with the
 164     original args as a fallback.
 165
 166     Based upon the value of the optional argument
 167     :code:`persist_at_shutdown` argument, (NEVER, IF_NOT_LOADED,
 168     ALWAYS), the :meth:`save` method of the class will be invoked just
 169     before program shutdown to give the class a chance to save its
 170     state somewhere.
 171
 172     .. note::
 173         The implementations of :meth:`save` and :meth:`load` and where the
 174         class persists its state are details left to the :class:`Persistent`
 175         implementation.  Essentially this decorator just handles the
 176         plumbing of calling your save/load and appropriate times and
 177         creates a transparent global singleton whose state can be
 178         persisted between runs.
 179
 180     """
 181
 182     def __init__(
 183         self,
 184         *,
 185         persist_at_shutdown: PersistAtShutdown = PersistAtShutdown.IF_NOT_LOADED,
 186     ):
 187         self.persist_at_shutdown = persist_at_shutdown
 188         self.instance = None
 189
 190     def __call__(self, cls: Persistent):
 191         @functools.wraps(cls)  # type: ignore
 192         def _load(*args, **kwargs):
 193
 194             # If class has already been loaded, act like a singleton
 195             # and return a reference to the one and only instance in
 196             # memory.
 197             if self.instance is not None:
 198                 logger.debug(
 199                     'Returning already instantiated singleton instance of %s.', cls.__name__
 200                 )
 201                 return self.instance
 202
 203             # Otherwise, try to load it from persisted state.
 204             was_loaded = False
 205             logger.debug('Attempting to load %s from persisted state.', cls.__name__)
 206             self.instance = cls.load()
 207             if not self.instance:
 208                 msg = 'Loading from cache failed.'
 209                 logger.warning(msg)
 210                 logger.debug('Attempting to instantiate %s directly.', cls.__name__)
 211                 self.instance = cls(*args, **kwargs)
 212             else:
 213                 logger.debug('Class %s was loaded from persisted state successfully.', cls.__name__)
 214                 was_loaded = True
 215
 216             assert self.instance is not None
 217
 218             if self.persist_at_shutdown is PersistAtShutdown.ALWAYS or (
 219                 not was_loaded and self.persist_at_shutdown is PersistAtShutdown.IF_NOT_LOADED
 220             ):
 221                 logger.debug('Scheduling a deferred called to save at process shutdown time.')
 222                 atexit.register(self.instance.save)
 223             return self.instance
 224
 225         return _load
 226
 227
 228 if __name__ == '__main__':
 229     import doctest
 230
 231     doctest.testmod()