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