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     now = datetime.datetime.now()
  68     return mtime.month == now.month and mtime.day == now.day and mtime.year == now.year
  69
  70
  71 def was_file_written_within_n_seconds(
  72     filename: str,
  73     limit_seconds: int,
  74 ) -> bool:
  75     """Returns True if filename was written within the pas limit_seconds
  76     seconds.
  77
  78     """
  79     if not file_utils.does_file_exist(filename):
  80         return False
  81
  82     mtime = file_utils.get_file_mtime_as_datetime(filename)
  83     now = datetime.datetime.now()
  84     return (now - mtime).total_seconds() <= limit_seconds
  85
  86
  87 class PersistAtShutdown(enum.Enum):
  88     """
  89     An enum to describe the conditions under which state is persisted
  90     to disk.  See details below.
  91
  92     """
  93
  94     NEVER = (0,)
  95     IF_NOT_LOADED = (1,)
  96     ALWAYS = (2,)
  97
  98
  99 class persistent_autoloaded_singleton(object):
 100     """A decorator that can be applied to a Persistent subclass (i.e.  a
 101     class with a save() and load() method.  It will intercept attempts
 102     to instantiate the class via it's c'tor and, instead, invoke the
 103     class' load() method to give it a chance to read state from
 104     somewhere persistent.
 105
 106     If load() fails (returns None), the c'tor is invoked with the
 107     original args as a fallback.
 108
 109     Based upon the value of the optional argument persist_at_shutdown,
 110     (NEVER, IF_NOT_LOADED, ALWAYS), the save() method of the class will
 111     be invoked just before program shutdown to give the class a chance
 112     to save its state somewhere.
 113
 114     The implementations of save() and load() and where the class
 115     persists its state are details left to the Persistent
 116     implementation.
 117
 118     """
 119
 120     def __init__(
 121         self,
 122         *,
 123         persist_at_shutdown: PersistAtShutdown = PersistAtShutdown.IF_NOT_LOADED,
 124     ):
 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(
 152                     f'Class {cls.__name__} was loaded from persisted state successfully.'
 153                 )
 154                 was_loaded = True
 155
 156             assert self.instance is not None
 157
 158             if self.persist_at_shutdown is PersistAtShutdown.ALWAYS or (
 159                 not was_loaded
 160                 and self.persist_at_shutdown is PersistAtShutdown.IF_NOT_LOADED
 161             ):
 162                 logger.debug(
 163                     'Scheduling a deferred called to save at process shutdown time.'
 164                 )
 165                 atexit.register(self.instance.save)
 166             return self.instance
 167
 168         return _load