persistent.py

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