persistent.py

   1 #!/usr/bin/env python3
   2
   3 from abc import ABC, abstractmethod
   4 import atexit
   5 import functools
   6 import logging
   7
   8 import dill
   9
  10 import file_utils
  11
  12
  13 logger = logging.getLogger(__name__)
  14
  15
  16 class Persistent(ABC):
  17     @abstractmethod
  18     def save(self):
  19         pass
  20
  21     @abstractmethod
  22     def load(self):
  23         pass
  24
  25
  26 class persistent_autoload_singleton(Persistent):
  27     def __init__(self, filename: str, *, max_age_sec: int = 0):
  28         self.filename = filename
  29         self.max_age_sec = max_age_sec
  30         self.instance = None
  31
  32     def __call__(self, cls):
  33         @functools.wraps(cls)
  34         def _load(*args, **kwargs):
  35
  36             # If class has already been loaded, act like a singleton
  37             # and return a reference to the one and only instance in
  38             # memory.
  39             if self.instance is not None:
  40                 logger.debug(
  41                     f'Returning already instantiated singleton instance of {cls.__name__}.'
  42                 )
  43                 return self.instance
  44
  45             if not self.load():
  46                 assert self.instance is None
  47                 logger.debug(
  48                     f'Instantiating {cls.__name__} directly.'
  49                 )
  50                 self.instance = cls(*args, **kwargs)
  51
  52             # On program exit, save state to disk.
  53             atexit.register(self.save)
  54             assert self.instance is not None
  55             return self.instance
  56         return _load
  57
  58     def load(self) -> bool:
  59         if (
  60                 file_utils.does_file_exist(self.filename)
  61                 and (
  62                     self.max_age_sec == 0 or
  63                     file_utils.get_file_mtime_age_seconds(self.filename) <= self.max_age_sec
  64                 )
  65         ):
  66             logger.debug(
  67                 f'Attempting to load from file {self.filename}'
  68             )
  69             try:
  70                 with open(self.filename, 'rb') as f:
  71                     self.instance = dill.load(f)
  72                     return True
  73             except Exception:
  74                 self.instance = None
  75                 return False
  76         return False
  77
  78     def save(self) -> bool:
  79         if self.instance is not None:
  80             logger.debug(
  81                 f'Attempting to save {type(self.instance).__name__} to file {self.filename}'
  82             )
  83             try:
  84                 with open(self.filename, 'wb') as f:
  85                     dill.dump(self.instance, f, dill.HIGHEST_PROTOCOL)
  86                 return True
  87             except Exception:
  88                 return False
  89         return False