persistent.py

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