From: Scott Gasch Date: Wed, 22 Sep 2021 01:25:50 +0000 (-0700) Subject: Persistent state. X-Git-Url: https://wannabe.guru.org/gitweb/?a=commitdiff_plain;h=c46018e2b1ddc78f9df557c3fb24d2c2c849f054;p=python_utils.git Persistent state. --- diff --git a/persistent.py b/persistent.py new file mode 100644 index 0000000..30e4ccb --- /dev/null +++ b/persistent.py @@ -0,0 +1,89 @@ +#!/usr/bin/env python3 + +from abc import ABC, abstractmethod +import atexit +import functools +import logging + +import dill + +import file_utils + + +logger = logging.getLogger(__name__) + + +class Persistent(ABC): + @abstractmethod + def save(self): + pass + + @abstractmethod + def load(self): + pass + + +class persistent_autoload_singleton(Persistent): + def __init__(self, filename: str, *, max_age_sec: int = 0): + self.filename = filename + self.max_age_sec = max_age_sec + self.instance = None + + def __call__(self, cls): + @functools.wraps(cls) + def _load(*args, **kwargs): + + # If class has already been loaded, act like a singleton + # and return a reference to the one and only instance in + # memory. + if self.instance is not None: + logger.debug( + f'Returning already instantiated singleton instance of {cls.__name__}.' + ) + return self.instance + + if not self.load(): + assert self.instance is None + logger.debug( + f'Instantiating {cls.__name__} directly.' + ) + self.instance = cls(*args, **kwargs) + + # On program exit, save state to disk. + atexit.register(self.save) + assert self.instance is not None + return self.instance + return _load + + def load(self) -> bool: + if ( + file_utils.does_file_exist(self.filename) + and ( + self.max_age_sec == 0 or + file_utils.get_file_mtime_age_seconds(self.filename) <= self.max_age_sec + ) + ): + logger.debug( + f'Attempting to load from file {self.filename}' + ) + try: + with open(self.filename, 'rb') as f: + self.instance = dill.load(f) + return True + except Exception: + self.instance = None + return False + return False + + def save(self) -> bool: + if self.instance is not None: + logger.debug( + f'Attempting to save {type(self.instance).__name__} to file {self.filename}' + ) + try: + with open(self.filename, 'wb') as f: + dill.dump(self.instance, f, dill.HIGHEST_PROTOCOL) + return True + except Exception: + return False + return False diff --git a/tests/decorator_utils_test.py b/tests/decorator_utils_test.py new file mode 100755 index 0000000..195dd63 --- /dev/null +++ b/tests/decorator_utils_test.py @@ -0,0 +1,24 @@ +#!/usr/bin/env python3 + +import unittest + +import decorator_utils as du + +import unittest_utils as uu + + +class TestDecorators(unittest.TestCase): + + def test_singleton(self): + + @du.singleton + class FooBar(): + pass + + x = FooBar() + y = FooBar() + self.assertTrue(x is y) + + +if __name__ == '__main__': + unittest.main()