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
  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     assert mtime
  68     now = datetime.datetime.now()
  69     return mtime.month == now.month and mtime.day == now.day and mtime.year == now.year
  70
  71
  72 def was_file_written_within_n_seconds(
  73     filename: str,
  74     limit_seconds: int,
  75 ) -> bool:
  76     """Returns True if filename was written within the pas limit_seconds
  77     seconds.
  78
  79     """
  80     if not file_utils.does_file_exist(filename):
  81         return False
  82
  83     mtime = file_utils.get_file_mtime_as_datetime(filename)
  84     assert mtime
  85     now = datetime.datetime.now()
  86     return (now - mtime).total_seconds() <= limit_seconds
  87
  88
  89 class PersistAtShutdown(enum.Enum):
  90     """
  91     An enum to describe the conditions under which state is persisted
  92     to disk.  See details below.
  93
  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
 122     def __init__(
 123         self,
 124         *,
 125         persist_at_shutdown: PersistAtShutdown = PersistAtShutdown.IF_NOT_LOADED,
 126     ):
 127         self.persist_at_shutdown = persist_at_shutdown
 128         self.instance = None
 129
 130     def __call__(self, cls: Persistent):
 131         @functools.wraps(cls)  # type: ignore
 132         def _load(*args, **kwargs):
 133
 134             # If class has already been loaded, act like a singleton
 135             # and return a reference to the one and only instance in
 136             # memory.
 137             if self.instance is not None:
 138                 logger.debug(
 139                     f'Returning already instantiated singleton instance of {cls.__name__}.'
 140                 )
 141                 return self.instance
 142
 143             # Otherwise, try to load it from persisted state.
 144             was_loaded = False
 145             logger.debug(f'Attempting to load {cls.__name__} from persisted state.')
 146             self.instance = cls.load()
 147             if not self.instance:
 148                 msg = 'Loading from cache failed.'
 149                 logger.warning(msg)
 150                 logger.debug(f'Attempting to instantiate {cls.__name__} directly.')
 151                 self.instance = cls(*args, **kwargs)
 152             else:
 153                 logger.debug(
 154                     f'Class {cls.__name__} was loaded from persisted state successfully.'
 155                 )
 156                 was_loaded = True
 157
 158             assert self.instance is not None
 159
 160             if self.persist_at_shutdown is PersistAtShutdown.ALWAYS or (
 161                 not was_loaded
 162                 and self.persist_at_shutdown is PersistAtShutdown.IF_NOT_LOADED
 163             ):
 164                 logger.debug(
 165                     'Scheduling a deferred called to save at process shutdown time.'
 166                 )
 167                 atexit.register(self.instance.save)
 168             return self.instance
 169
 170         return _load