persistent.py

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