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         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                 logger.debug(f'Attempting to instantiate {cls.__name__} directly.')
 150                 self.instance = cls(*args, **kwargs)
 151             else:
 152                 logger.debug(
 153                     f'Class {cls.__name__} was loaded from persisted state successfully.'
 154                 )
 155                 was_loaded = True
 156
 157             assert self.instance is not None
 158
 159             if self.persist_at_shutdown is PersistAtShutdown.ALWAYS or (
 160                 not was_loaded
 161                 and self.persist_at_shutdown is PersistAtShutdown.IF_NOT_LOADED
 162             ):
 163                 logger.debug(
 164                     'Scheduling a deferred called to save at process shutdown time.'
 165                 )
 166                 atexit.register(self.instance.save)
 167             return self.instance
 168
 169         return _load