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